사이드 프로젝트[001]. Pyspark를 이용한 Write/Read

1. 구성환경 : 이전 글과 동일 + jupyter notebook 사용

 

2. 코드

• Spark와 세션 연결

import warnings
warnings.filterwarnings(action='ignore') 
from pyspark.sql import SparkSession

input_uri = "mongodb://hyeongju:dbgudwn1!@127.0.0.1:27017/mydatabase.testcol?authSource=admin"
output_uri = "mongodb://hyeongju:dbgudwn1!@127.0.0.1:27017/mydatabase.testcol?authSource=admin"

myspark = SparkSession\
        .builder\
        .appName("twitter")\
        .config("spark.mongodb.input.uri", input_uri)\
        .config("spark.mongodb.output.uri", output_uri)\
        .config('spark.jars.packages','org.mongodb.spark:mongo-spark-connector_2.12:2.4.2')\
        .getOrCreate()

• 'testcol' 이라는 Collection에 데이터 wrtie

people = myspark.createDataFrame([("Bilbo Baggins",  50), ("Gandalf", 1000)
, ("Thorin", 195), ("Balin", 178), ("Kili", 77),("Dwalin", 169), ("Oin", 167)
,("Gloin", 158), ("Fili", 82), ("Bombur", None)], ["name", "age"])

people.write.format("mongo").mode("append").save()

people.show()

write 결과

 

• 저장된 데이터 로드

df = myspark.read.format("mongo").load()
df.printSchema()

testcol 저장된 데이터의 구성

df.show(5)
#값을 추가하면 row 5줄만 출력한다.

df.show(5) 결과