<jupyter_start><jupyter_text>스파크 스트리밍 실습 > 파일을 읽어서 파이썬 스트리밍 실습을 합니다 목표1. 특정 경로에 존재하는 Json 파일을 스트리밍으로 읽는 실습을 수행합니다1. 스트리밍 집계 함수를 실습합니다1. 스트리밍 변환 함수를 실습합니다1. 여러 스트림을 조인하는 연산을 실습합니다1. 이벤트 시간 기준의 텀블링 윈도우 처리 실습1. 워터마크를 이용하여 1분 단위로 접속한 이용자의 수를 가지는 테이블을 구성하고 조회합니다1. UDF 를 이용하여, 워터마크와 윈도우 함수를 이용하여 누적 최고 매출 이용자를 가지는 테이블을 구성하고 조회합니다1. DataFrame 을 이용하여, 동일한 기능을 실습합니다<jupyter_code>from pyspark.sql import SparkSession
from pyspark.sql.functions import *

spark = SparkSession.builder.appName("foo-bar") \
    .config("spark.sql.session.timeZone", "Asia/Seoul") \
    .config("spark.jars.packages", "io.delta:delta-core_2.12:0.7.0") \
    .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
    .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog") \
    .getOrCreate()<jupyter_output><empty_output><jupyter_text>스파크 스트리밍 처리에서 자주 활용되는 함수들<jupyter_code>from random import uniform
from time import sleep
from IPython.display import display, clear_output
# 스트림 테이블을 주기적으로 조회하는 함수
def displayStream(name, sql, iterations, sleep_secs):
    i = 1
    for x in range(iterations):
        clear_output(wait=True)
        display('[' + name + '] Iteration: '+str(i)+', Score: '+str(uniform(0, 1)))
        spark.sql(sql).show(truncate=False)
        sleep(sleep_secs)    
        i += 1
from IPython.display import JSON
# 해당 스트리밍 쿼리의 마지막 상태를 확인하는 함수
def checkLastProgress(query):
    JSON(query.lastProgress)
# 해당 스트리밍 쿼리를 재실행 시에, 이미 시작된 쿼리를 사전에 종료하기 위한 함수
def stopQueryIfStarted(query):
    try:
        query.stop()
    except:
        pass
# 테이블 목록 출력
def showTables():
    spark.sql("show tables").show()<jupyter_output><empty_output><jupyter_text>1. 특정 경로에 존재하는 Json 파일을 읽어서 스트리밍 파이프라인을 구성합니다<jupyter_code>static = spark.read.json("data/activity-data")
dataSchema = static.schema
static.printSchema()
static.show(2, truncate=False)
streaming = spark.readStream.schema(dataSchema) \
    .option("maxFilesPerTrigger", 1).json("data/activity-data")
spark.conf.set("spark.sql.shuffle.partitions", 5)  # 로컬 모드에서는 너무 많은 파티션은 오히려 성능을 떨어뜨리므로 셔플 수를 줄입니다<jupyter_output><empty_output><jupyter_text>2. 스트리밍 집계 함수를 실습합니다<jupyter_code>activityCounts = streaming.groupBy("gt").count()  # 스트리밍은 바로 출력 대신에 다른 스트리밍 싱크에 보내기 위한 로직만 먼저 작성합니다

# 아래와 같이 "activity_counts" 라는 임의의 "memory" 테이블에 매번 모든 결과를 "complete" 하게 전송하는 activityQuery 를 생성합니다
activityQuery = activityCounts.writeStream \
    .queryName("activity_counts") \
    .format("memory") \
    .outputMode("complete") \
    .start()

# 상태를 확인하기 위해서 항상 스트리밍 쿼리를 별도의 객체로 만들어 두는 것이 편리합니다.
activityQuery.status
# 위와 같이 start 한 이후, 백그라운드에서 코드를 수행시키기 위해서 아래와 같이 실행합니다
DEBUG = False
if DEBUG:
    activityQuery.id  # get the unique identifier of the running query that persists across restarts from checkpoint data

    activityQuery.runId       # get the unique id of this run of the query, which will be generated at every start/restart

    activityQuery.name        # get the name of the auto-generated or user-specified name

    activityQuery.explain()   # print detailed explanations of the query

    activityQuery.stop()      # stop the query

    activityQuery.awaitTermination()   # block until query is terminated, with stop() or with error

    activityQuery.exception       # the exception if the query has been terminated with error

    activityQuery.recentProgress  # an array of the most recent progress updates for this query

    activityQuery.lastProgress    # the most recent progress update of this streaming query<jupyter_output><empty_output><jupyter_text>현재 활성화된 스트리밍 쿼리에 대한 확인 및 조회<jupyter_code>spark.streams.active  # get the list of currently active streaming queries

# spark.streams.get(id)  # get a query object by its unique id

# 라이브 어플리케이션의 경우는 백그라운드에서 항상 동작해야 하므로, 아래와 같이 어플리케이션이 종료될 때까지 대기하는 명령을 수행해야 합니다.
# spark.streams.awaitAnyTermination()  # block until any one of them terminates
displayStream("activity_counts", "select * from activity_counts", 3, 1)
displayStream("Counts of Activities", "select * from activity_counts", 3, 1)<jupyter_output><empty_output><jupyter_text>3. 스트리밍 변환 함수를 실습합니다<jupyter_code>simpleTransform = streaming.withColumn("stairs", expr("gt like '%stairs%'")) \
    .where("stairs") \
    .where("gt is not null") \
    .select("gt", "model", "arrival_time", "creation_time") \
    .writeStream \
    .queryName("simple_transform") \
    .format("memory") \
    .outputMode("append") \
    .start()
displayStream("Counts of Activities", "select * from simple_transform", 3, 1)
checkLastProgress(activityQuery)
showTables()<jupyter_output>+--------+----------------+-----------+
|database|       tableName|isTemporary|
+--------+----------------+-----------+
|        | activity_counts|       true|
|        |simple_transform|       true|
+--------+----------------+-----------+<jupyter_text>4. 여러 스트림을 조인하는 연산을 실습합니다<jupyter_code># 아래와 같이 데이터소스를 처음에 읽어서 해당 데이터프레임을 만들어두면, 해당 스트림을 사용하면 파일을 처음부터 읽어서 테스트가 용이합니다
deviceModelStats = None
stopQueryIfStarted(deviceModelStats)
historicalAgg = static.groupBy("gt", "model").avg()
deviceModelStats = streaming.drop("Arrival_Time", "Creation_Time", "Index") \
    .cube("gt", "model").avg() \
    .join(historicalAgg, ["gt", "model"]) \
    .writeStream \
    .queryName("device_counts") \
    .format("memory") \
    .outputMode("complete") \
    .start()
displayStream("device_counts", "select * from device_counts", 5, 1)<jupyter_output><empty_output><jupyter_text>5. 이벤트 시간 기준의 텀블링 윈도우 처리 실습<jupyter_code>withEventTime = streaming.selectExpr("*", "cast(cast(Creation_Time as double)/1000000000 as timestamp) as event_time")
eventPerWindow = None
stopQueryIfStarted(eventPerWindow)
eventPerWindow = withEventTime.groupBy(window(col("event_time"), "10 minutes")).count() \
     .writeStream \
     .queryName("pyevents_per_window") \
     .format("memory") \
     .outputMode("complete") \
     .start()
displayStream("pyevents_per_window", "select * from pyevents_per_window", 5, 3)
eventPerUserWindow = None
stopQueryIfStarted(eventPerUserWindow)
eventPerUserWindow = withEventTime.groupBy(window(col("event_time"), "10 minutes"), "User").count() \
     .writeStream \
     .queryName("pyevents_per_user_window") \
     .format("memory") \
     .outputMode("complete") \
     .start()
displayStream("pyevents_per_user_window", "select * from pyevents_per_user_window order by count desc limit 5", 5, 1)
static.show(10, truncate=False)
static.groupBy("Device", "User").count().sort(desc("count")).limit(5).show()
userRank = None
stopQueryIfStarted(userRank)

# 해당 그룹 파티션 내에서 X, Y, Z 역순 정렬 후에, 유일한 번호를 매겨서 출력
userRank = withEventTime \
    .withWatermark("event_time", "10 minutes") \
    .groupBy("User", window(col("event_time"), "10 minutes")).count() \
    .writeStream \
    .queryName("user_rank") \
    .format("memory") \
    .outputMode("complete") \
    .start()
displayStream("user_rank", "select * from user_rank order by window desc, User, count desc", 1, 1)<jupyter_output><empty_output><jupyter_text>5. 워터마크를 이용하여 1분 단위로 접속한 이용자의 수를 가지는 테이블을 구성하고 조회합니다<jupyter_code>df = spark.read.option("header", "true").csv("data/tbl_user.csv")
df.printSchema()
df.show()
df.rdd.zipWithIndex().collect()
df.withColumn("id", monotonically_increasing_id() + 1).show()
df = spark.sql("select 'a,b' as col1")
df2 = df.withColumn("codes" , explode(split("col1" , ","))).drop("col1")
df2.show()
from pyspark.sql.types import StructType,StructField, StringType, IntegerType

data = [("James","","Smith","36636","M",3000),
    ("Michael","Rose","","40288","M",4000),
    ("Robert","","Williams","42114","M",4000),
    ("Maria","Anne","Jones","39192","F",4000),
    ("Jen","Mary","Brown","","F",-1)
  ]

schema = StructType([ \
    StructField("firstname",StringType(),True), \
    StructField("middlename",StringType(),True), \
    StructField("lastname",StringType(),True), \
    StructField("id", StringType(), True), \
    StructField("gender", StringType(), True), \
    StructField("salary", IntegerType(), True) \
  ])
 
df = spark.createDataFrame(data=data,schema=schema)
df.printSchema()
df.show(truncate=False)
structureData = [
    (("James","","Smith"),"36636","M",3100),
    (("Michael","Rose",""),"40288","M",4300),
    (("Robert","","Williams"),"42114","M",1400),
    (("Maria","Anne","Jones"),"39192","F",5500),
    (("Jen","Mary","Brown"),"","F",-1)
  ]
structureSchema = StructType([
        StructField('name', StructType([
             StructField('firstname', StringType(), True),
             StructField('middlename', StringType(), True),
             StructField('lastname', StringType(), True)
             ])),
         StructField('id', StringType(), True),
         StructField('gender', StringType(), True),
         StructField('salary', IntegerType(), True)
         ])

df2 = spark.createDataFrame(data=structureData,schema=structureSchema)
df2.printSchema()
df2.show(truncate=False)
df2.selectExpr("name.firstname as first", "name.middlename", "name.lastname").show()<jupyter_output>+-------+----------+--------+
|  first|middlename|lastname|
+-------+----------+--------+
|  James|          |   Smith|
|Michael|      Rose|        |
| Robert|          |Williams|
|  Maria|      Anne|   Jones|
|    Jen|      Mary|   Brown|
+-------+----------+--------+
{'fraction_non_alphanumeric': 0.16186468112657817, 'fraction_numerical': 0.0160785583252401, 'mean_word_length': 3.9141039236479322, 'pattern_counts': {'":': 0, '<': 35, '<?xml version=': 0, '>': 36, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 1, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '9', 'dir': 'jupyter-structured-clean-dedup', 'id': '501845', 'license': 'Apache-2.0', 'n_tokens_mistral': 3887, 'n_tokens_neox': 4235, 'n_words': 734, 'path': 'spark/notebooks/pyspark-stream-1.ipynb', 'repo_name': 'sw-woo/data-engineer-training', 'size': '9267'}