从kafka服务器接收JSON后，更改Pyspark数据框架的两列的数据类型，但正在获取null值
我希望更改我构建的pyspark数据框架的“ Tweet_id”和“用户ID”。我希望两者都具有类型整数的数据类型。 我的代码在下面... import findspark from p…
在火花结构化流应用程序中计算Kafka滞后
我正在尝试计算我的火花结构流应用程序上的kafka滞后。 我可以从随附的实际数据中获得当前处理的偏移量。 有没有办法通过SPARK界面以编程方式在Kafka…
火花结构化流databrick上没有控制台输出
我正在尝试将带有套接字的Databrick中的结构化流作为源，并将控制器作为输出接收器。 但是，我无法在Databrick上看到任何输出。 from pyspark.sql.fun…
在阅读时，使用Spark写给Kafka的JSON字符串未正确转换给Kafka
我读了一个.csv文件以创建数据框架，我想将数据写入Kafka主题。代码是以下内容 df = spark.read.format("csv").option("header", "true").load(f'{fil…
火花结构化流和Neo4J
我的目标是使用Spark结构化流从MongoDB集合中将转换的数据从MongoDB集合写入Neo4J。根据NEO4J文档的说法，“ neo4j连接器，适用于Apache Spark “版本…
火花结构化流 - Pyspark和Scala之间的水印行为
我正在尝试使用Watermark在Python中部署一个结构化的流媒体应用程序。我使用控制台接收器来测试这一点，但发现了Pyspark和Scala之间的怪异行为。我将…
结构化流 - 从gke上读取strimzi kafka，每10分钟将数据写入mongo
我在KAFKA主题中有数据（数据每10分钟发布一次），我计划使用Apache Spark结构化流（批处理模式）读取此数据，然后将其推入MongoDB。 请注意： 这将在…
使用Abris for Spark的Confluent Magic Bytes自动解决模式
是否有一种方法可以通过针对每个消息的领先魔术字节自动解决模式，其中包含该消息的架构ID？， 我们知道， abris添加了此ID （魔术字节）编码架构以支…
火花结构化流（批处理模式） - 同时运行依赖作业
我有一个在GCP DataProc群集上运行的结构化流媒体程序，该程序每10分钟读取KAFKA的数据，然后进行处理。 这是一个多租户系统，即该程序将读取来自多个…
如何将Pyspark DataFrame发送到Kafka主题？
Pyspark版本-2.4.7 KAFKA版本-2.13_3.2.0 HI，我是Pyspark和流媒体属性的新手。我在Internet中遇到了一些资源，但我仍然无法弄清楚如何将Pyspark数据…
如何在foreachBatch函数中打印/日志输出？
使用表流，我正在尝试使用ForeachBatch df.writestream .format("delta") .foreachBatch(WriteStreamToDelta) ... WritestreamTodelta编写流，看起来 …
如何将KAFKA的单个主题流式传输，通过键过滤到HDF的多个位置？
我不是要在多个HDFS位置上流式传输数据，该位置通过键过滤。因此，下面的代码不起作用。请帮助我找到编写此代码的正确方法 val ER_stream_V1 = spark …
Writestream在需要客户端证书的Azure API管理背后的Elasticsearch的数据框架？
我们有一个环境，我们的Elasticsearch在Azure API管理背后受到保护。我们将其锁定在客户证书要求（以及其他安全措施）中。拒绝客户证书进入APIM的呼叫…
在火花中使用Pubsub Lite库获取错误
我在使用Spark结构化流向GCP PubSub Lite发布消息时会遇到错误。 我无法在Spark中使用Writestream，因为我想在Spark中的ForeachBatch接收器中使用它，…
GCP Composer 2（气流2）数据Proc Operator-将软件包传递给PYSPARK_JOB
我正在使用GCP Composer2安排Pyspark（结构化流）作业， Pyspark代码读/写入Kafka。 DAG使用运算符 - dataproccreateclusteroperator （创建一个GKE群…
