Spark Structred Streaming Pyspark Sink Csv не добавляет - PullRequest
Новая
       63

Spark Structred Streaming Pyspark Sink Csv не добавляет

0 голосов
/ 11 января 2020

Напишите json в Kafka Topi c и прочитайте json из kafka topi c. На самом деле я подписываюсь topi c и пишу консоль построчно. Но я должен утопить / записать файл csv. Но я не могу. Я пишу CSV один раз, но не добавляю.

Вы можете увидеть мой код ниже.

Спасибо! 

import pyspark
from pyspark.sql import SparkSession
from pyspark.sql.types import *
import pyspark.sql.functions as func
spark = SparkSession.builder\
                    .config('spark.jars.packages', 'org.apache.spark:spark-sql-kafka-0-10_2.11:2.3.0') \
                    .appName('kafka_stream_test')\
                    .getOrCreate()
ordersSchema = StructType() \
        .add("a", StringType()) \
        .add("b", StringType()) \
        .add("c", StringType()) \
        .add("d", StringType())\
        .add("e", StringType())\
        .add("f", StringType())

df = spark \
    .readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "localhost:9092") \
    .option("subscribe", "product-views") \
    .load()\


df_query = df \
    .selectExpr("cast(value as string)") \
    .select(func.from_json(func.col("value").cast("string"),ordersSchema).alias("parsed"))\
    .select("parsed.a","parsed.b","parsed.c","parsed.d","parsed.e","parsed.f")\


df = df_query \
    .writeStream \
    .format("csv")\
    .trigger(processingTime = "5 seconds")\
    .option("path", "/var/kafka_stream_test_out/")\
    .option("checkpointLocation", "/user/kafka_stream_test_out/chk") \
    .start()

df.awaitTermination()

1 Ответ

0 голосов
/ 11 января 2020

Да, потому что вам нужна эта дополнительная опция .option("format", "append"): 

aa = df_query \
    .writeStream \
    .format("csv")\
    .option("format", "append")\
    .trigger(processingTime = "5 seconds")\
    .option("path", "/var/kafka_stream_test_out/")\
    .option("checkpointLocation", "/user/kafka_stream_test_out/chk") \
    .outputMode("append") \
    .start()
...