如何正确使用Pyspark将数据发送到Kafka Broker？
我想编写一个简单的pyspark作业，从kafka代理主题接收数据，对该数据进行一些转换，并将转换后的数据放入不同的kafka代理主题。 我有以下代码，它从kafka主题读取数据，但在运行sendkafka函数时没有效果： from pyspark import SparkConf, Sp...
Spark Streaming + Kafka: Spark异常：无法找到Set的Leader Offsets
我正在尝试设置Spark Streaming来获取Kafka队列中的消息。我遇到了以下错误： py4j.protocol.Py4JJavaError: An error occurred while calling o30.createDirectStream. : org.apache.s...
为什么启动StreamingContext会失败并显示“IllegalArgumentException: requirement failed: No output operations registered, so nothing to execute”？
我正在尝试执行一个以 Twitter 为源的 Spark Streaming 示例，如下所示：public static void main (String.. args) { SparkConf conf = new SparkConf().setAppName("Spark_St...
使用Python的Spark：如何解决阶段x包含一个非常大（xxx KB）的任务。最大推荐任务大小为100 KB。
我刚刚创建了一个Python列表，其中包含range(1,100000)。 使用SparkContext执行了以下步骤：a = sc.parallelize([i for i in range(1, 100000)]) b = sc.parallelize([i for i in range...
Spark的ssc.textFileStream无法从目录流式传输任何文件
我正在尝试使用 Eclipse（具有 Maven 配置），使用 2 个工作器，每个工作器都有 2 个核心，或者也尝试使用 spark-submit 执行以下代码。public class StreamingWorkCount implements Serializable { pub...
Spark Streaming历史状态
我正在构建实时处理来检测欺诈ATM卡交易。为了有效地检测欺诈，逻辑需要按卡片分组的最后交易日期和每天（或过去24小时）交易金额总和。 其中一个用例是，如果卡片在本国以外的交易距离上次在该国交易超过30天，则发送警报以可能是欺诈。 因此，我尝试使用Spark流作为解决方案。为了实现这一点（可...
检查Spark流式处理作业是否挂起的最佳方法
我有一个Spark流应用程序，基本上从Kafka接收触发消息，启动可能需要长达2小时的批处理。 曾经出现过一些作业无限期挂起并且没有在通常时间内完成的情况，目前我们无法找到作业状态的方法，除非手动检查Spark UI。我希望有一种方式可以确定当前正在运行的Spark作业是否挂起。因此，如果它...
SparkStreaming：避免checkpointLocation检查
我正在编写一个库，将Apache Spark与自定义环境集成。我正在实现自定义的流数据源和流写入器。 我正在开发的一些数据源在应用程序崩溃后不可恢复。如果应用程序重新启动，它需要重新加载所有数据。因此，我们希望避免用户显式设置“checkpointLocation”选项。 但是如果未提供该选...
在集群模式下使用spark-submit共享配置文件
在开发期间，我一直在以"client"模式运行我的Spark作业。我使用"--file"与执行器共享配置文件。驱动程序是在本地读取配置文件的。现在我想在"cluster"模式下部署作业，但我现在难以与驱动程序共享配置文件。 例如，我将配置文件名作为extraJavaOptions传递给驱动程...
如何将数据从Kafka传输到Spark Streaming？
我试图从kafka传递数据到spark streaming。到目前为止，我已经完成以下步骤：安装了kafka和spark使用默认属性配置启动zookeeper使用默认属性配置启动kafka服务器启动kafka生产者启动kafka消费者从生产者发送消息到消费者。 运行良好。编写了kafka-sp...