# Schema 정의
schema = StructType([
    StructField("eqp_id", StringType()),
    StructField("single_no", StringType()),
    StructField("img_nm", StringType()),
    StructField("img_path", StringType()),
    StructField("upload_dttm", StringType())
])

# Kafka 데이터 수신을 위한 DataFrame 생성
kafka_df = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", BOOTSTRAP_SERVERS) \
    .option("subscribe", TOPIC_NAME) \
    .option("startingOffsets", "earliest") \
    .load()

# json 데이터 파싱
parsed_df = kafka_df.selectExpr("CAST(value AS STRING)") \
    .select(from_json("value", schema).alias("data")) \
    .select("data.*")