pyspark - jjin-choi/study_note GitHub Wiki

pyspark 에서 duplicate 없애는 방법

• https://www.datasciencemadesimple.com/distinct-value-of-dataframe-in-pyspark-drop-duplicates/

pyspark 에서 여러 DB 접속하는 방법

from pyspark.sql import SparkSession
 
# mongo DB              
spark = SparkSession.builder
                    .appName('mongoDB')
                    .config('spark.mongodb.input.uri','mongodb://10.230.74.24:27020/dvlr.sim_log')
                    .config('spark.mongodb.output.uri','mongodb://10.230.74.24:27020/dvlr.sim_log')
                    .getOrCreate()
df = spark.read.format('mongo')
               .option('uri','mongodb://10.230.74.24:27020/dvlr.sim_log')
               .load()

# postgreSQL
spark = SparkSession.builder
                    .appName('Pyspark connected with Postgre')
                    .config('spark.jars', 'postgresql-42.2.23.jar')
                    .getOrCreate()
df = spark.read.format('jdbc')
               .option('url', 'jdbc:postgresql://10.230.74.162:5432/ibuilder')
               .option('dbtable', 'NRE_LICENSE')
               .option('user','jongwoo6969')
               .option('password','fbwhddn77^^')
               .option('driver', 'org.postgresql.Driver')
               .load()

pyspark function

• collect() : 해당 dataframe 의 모든 row 를 반환

import pandas as pd

# df : spark DataFrame
pddf = pd.DataFrame(df.collect())

• cache() : 자주 불리게 되는 dataframe 은 cache 라는 함수를 사용하여 메모리에 남겨둠