结构Spark流的scala控制微批量

kpbwa7wx 于 2021-06-04 发布在 Kafka

关注(0)|答案(1)|浏览(490)

我从一个kafka主题中读取数据，并以分区模式将其放入azureadls（类似hdfs）。
我的代码如下：

val df = spark
      .readStream
      .format("kafka")
      .option("kafka.bootstrap.servers", bootstrapServers)
      .option("subscribe", topic)
      .option("failOnDataLoss", false)
      .load()
      .selectExpr(/*"CAST(key AS STRING)",*/ "CAST(value AS STRING)").as(Encoders.STRING)
df.writeStream
      .partitionBy("year", "month", "day", "hour", "minute")
      .format("parquet")
      .option("path", outputDirectory)
      .option("checkpointLocation", checkpointDirectory)
      .outputMode("append")
      .start()
      .awaitTermination()

我有大约2000条记录/秒，我的问题是spark每45秒插入一次数据，我希望立即插入数据。
有人知道如何控制微批量的大小吗？

scala apache-kafka apache-spark spark-structured-streaming Azure

来源：https://stackoverflow.com/questions/56891194/control-micro-batch-of-structured-spark-streaming

1条答案

按热度按时间

pexxcrt21#

从spark 2.3版本开始，可以使用连续处理模式。在官方文件里。您可以看到，此模式仅支持三个接收器，并且只有kafka接收器可用于生产，并且“最好使用kafka作为源和接收器来观察端到端的低延迟处理”

df
.writeStream
.format("kafka")
.option("kafka.bootstrap.servers", "localhost:9092")
.option("checkpointLocation", "/tmp/0")
.option("topic", "output0")
.trigger(Trigger.Continuous("0 seconds"))
.start()

所以，现在看来，你可以´t使用连续模式将hdfs用作接收器。在你的情况下，也许你可以测试阿克卡流和阿尔帕卡连接器

赞(0）回复(0）举报 2021-06-05

我来回答

结构Spark流的scala控制微批量

1条答案

相关问题

热门标签

最新问答