Spark (Kafka to Hadoop)-pyspark

 

 

개발 서버에 Spark를 사용한 아래와 같은 파이프라인 demo를 구성하였습니다. 

Kafka에서 가져온 데이터를 Read하여 HDFS에 ORC나 Parquet 형식으로 저장합니다. 

Kafka to hadoop source

#contest.py from pyspark.sql import SparkSession sc = SparkSession.builder.getOrCreate() sc.sparkContext.setLogLevel('ERROR') # Read stream log = sc.readStream.format("kafka") \ .option("kafka.bootstrap.servers", "10.65.41.140:9092") \ .option("subscribe", "topic13") \ .option("startingOffsets", "earliest") \ .load() # Write stream - console query = log.selectExpr("CAST(value AS STRING)") \ .writeStream \ .format("console") \ .option("truncate", "false") \ .start() # Write stream - HDFS query2 = log.selectExpr("CAST(value AS STRING)") \ .writeStream \ .format("parquet") \ .outputMode("append") \ .option("checkpointLocation", "/check") \ .option("path", "/test") \ .start() query.awaitTermination() query2.awaitTermination()

 

실행 명령어
./bin/spark-submit --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.2.2,org.apache.kafka:kafka-clients:3.2.2,org.apache.spark:spark-streaming-kafka-0-10-assembly_2.12:3.2.2 contest.py

 

> Spark UI

 

> 기존 데이터 조회

 

 

> 실시간으로 가져오는 데이터 조회