apache-spark - 使用 Python 中的 Spark Structured Streaming 从 Kafka 读取数据并打印到控制台

Question

我有 kafka_2.13-2.7.0 在 Ubuntu 20.04 中。我运行 kafka 服务器和 zookeeper，然后创建一个主题并通过 nc -lk 9999 在其中发送一个文本文件.这个话题充满了数据。另外，我还有 spark-3.0.1-bin-hadoop2.7 在我的系统上。事实上，我想使用 kafka 主题作为 Spark Structured Streaming with python 的源。我的代码是这样的:

spark = SparkSession \
    .builder \
    .appName("APP") \
    .getOrCreate()

df = spark \
    .readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "localhost:9092") \
    .option("subscribe", "sparktest") \
    .option("startingOffsets", "earliest") \
    .load()

df.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
df.printSchema()

我通过 运行上面的代码spark-submit 用这个命令:

./spark-submit --packages org.apache.spark:spark-streaming-kafka-0-10_2.12:3.0.1,org.apache.spark:spark-sql-kafka-0-10_2.12:3.0.1 /home/spark/PycharmProjects/testSparkStream/KafkaToSpark.py 

代码运行没有任何异常，我收到这个输出，因为它在 Spark 站点中:

   root
    |-- key: binary (nullable = true)
    |-- value: binary (nullable = true)
    |-- topic: string (nullable = true)
    |-- partition: integer (nullable = true)
    |-- offset: long (nullable = true)
    |-- timestamp: timestamp (nullable = true)
    |-- timestampType: integer (nullable = true)

我的问题是 kafka 主题充满了数据；但是由于在输出中运行代码，没有任何数据。你能指导我这里有什么问题吗？

Answer 1

原样的代码不会打印出任何数据，而只会为您提供一次架构。
您可以按照一般 Structured Streaming Guide 中给出的说明进行操作。和 Structured Streaming + Kafka integration Guide查看如何将数据打印到控制台。请记住，在 Spark 中读取数据是一种惰性操作，没有操作(通常是 writeStream 操作)什么也做不了。
如果您对以下代码进行补充，您应该会看到所选数据(键和值)打印到控制台:

spark = SparkSession \
          .builder \
          .appName("APP") \
          .getOrCreate()

df = spark\
      .readStream \
      .format("kafka") \
      .option("kafka.bootstrap.servers", "localhost:9092") \
      .option("subscribe", "sparktest") \
      .option("startingOffsets", "earliest") \
      .load()
      

query = df.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)") \
    .writeStream \
    .format("console") \
    .option("checkpointLocation", "path/to/HDFS/dir") \
    .start()

query.awaitTermination()