spark-structured-streaming 相关问题

Spark Structured Streaming允许使用unifrom DataFrame和Dataset API处理实时数据流。

如何在pyspark的DataStreamReader中解析json字符串列并创建数据框

我正在读取来自kafka主题消息的消息DFRaw = spark.readStream \ .format（“kafka”）\ .option（“kafka.bootstrap.servers”，“localhost：9092”）\ ...

pyspark pyspark-sql spark-structured-streaming spark-streaming-kafka

回答 1 投票 0

有没有办法在火花流窗口中提取窗口开始时间和窗口结束时间？

我有一个DSTREAM，我使用窗口方法。然后我做其他操作，如reduceByKey。是否可以将窗口开始时间和结束时间添加到DSTREAM数据并将其用作...

spark-streaming spark-structured-streaming

回答 1 投票 0

[Structured Streaming]：将流数据帧写入Postgres

我有一个流数据帧，我试图写入数据库。有关于将rdd或df写入Postgres的文档。但是，我无法找到有关它如何的示例或文档......

postgresql apache-spark pyspark spark-structured-streaming

回答 1 投票 0

将流式JSON转换为DataFrame

问题：如何将JSON字符串转换为DataFrame并仅选择我想要的键？我上周刚开始使用Spark，我还在学习，所以请耐心等待。我正在使用Spark（2 ....

python-3.x apache-spark pyspark spark-structured-streaming

回答 1 投票 0

Spark结构化流：GroupStateTimeout的java.lang.NoClassDefFoundError [复制]

我正在尝试在https://spark.apache.org/docs/2.2.0/api/scala/index.html#org.apache.spark.sql.streaming.GroupState中定义的spark结构化流中使用mapGroupsWithState但是我明白了......

apache-spark spark-streaming spark-structured-streaming

回答 1 投票 0

为什么Spark应用程序失败并出现“ClassNotFoundException：找不到数据源：kafka”作为带有sbt程序集的uber-jar？

我正在尝试运行像https://github.com/apache/spark/blob/master/examples/src/main/scala/org/apache/spark/examples/sql/streaming/StructuredKafkaWordCount.scala这样的示例。我从Spark开始......

scala apache-spark sbt sbt-assembly spark-structured-streaming

回答 7 投票 19

使用Kafka通过Spark结构化批处理作业管理偏移

我有一个用例，我正在编写批处理作业，我需要阅读Kafka主题并将数据记录到HDFS。我的代码如下所示val df：DataFrame = spark.read .format（“kafka”）。option（“...

apache-spark apache-kafka apache-spark-sql spark-structured-streaming

回答 1 投票 -2

结构化流 - 从同一个流源连接2个DataFrame

我有这个Spark应用程序，它接收一个Twitter流。我添加了一个时间列：timestamp = datetime.datetime.fromtimestamp（time.time（））。strftime（'％Y-％m-％d％H：％M：％S'）timestamp_df = tmp_df2.withColumn（.. 。

python-3.x apache-spark window watermark spark-structured-streaming

回答 1 投票 0

登录spark结构化流/ SparkException：任务不可序列化

我正在尝试将Apache Flink应用程序（scala）移植到Spark结构化流媒体。该应用程序的基本工作是：从kafka读取消息做一些转换/处理输出零或更多...

scala apache-spark streaming apache-flink spark-structured-streaming

回答 1 投票 0

如何使用Spark Structured Streaming从目录读取读取时实现一次性处理？

我想使用流处理的概念从本地目录中读取文件，然后发布到Apache Kafka。我想过使用Spark Structured Streaming。怎么检查点......

apache-spark apache-kafka spark-structured-streaming

回答 1 投票 2

使用ForeachWriter在Spark流中实现Cassandra接收器

显然，Spark内置的Cassandra接收器没有内置支持。我在网上找到了这个例子，它基于...实现了Spark结构流的自定义Cassandra接收器。