Spark consumer(Kafka to Hadoop)

개발 서버에 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

기존 데이터 조회

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