使用火花结构化流Trigger.Once()计算策略运行多数据流查询，而不会出现应用程序故障。
浏览 38提问于2021-10-01得票数 0
我正在尝试使用Trigger.Once功能创建一个Spark Structured作业，以定期将新数据写入到Parquet数据湖中的CSV数据湖中。这就是我所拥有的： .readStream .csv("s3a://csv-data-lake-files") .writeStream
浏览 13提问于2017-08-16得票数 12
将外部包(jars)包含在火花放电外壳中的正确方法是什么？我想通过spark-sql-kafka库从卡夫卡中通过火花阅读，如下所述：。运行的机器上，还是在纱线运行的机器上？)我正在使用--master yarn和--deploy-mode client)，或者依赖于--packages
我不知道在pyspark-shell in PYSPARK_SUBMIT_ARGS之后指定的选项是否被排除在外(如果我试
浏览 0提问于2019-12-11得票数 0
我使用spark-structured-streaming作为消费者从kafka获取数据，按照指南参考https://spark.apache.org/docs/latest/structured-streaming-kafka-integration.html这是我的问题:程序运行良好，但一些容器很少失败(但它确实发生了)，导致了一些损坏的拼接文件。它将导致错误，如不是拼图文件(长度太小: 4)或[.
浏览 20提问于2019-05-25得票数 0
2回答
我在DataFrame中有3列：- time:TimeStamp，col1:Double，col2:Double我想执行以下操作：
dataFrame.withWatermark("time", "10
浏览 0提问于2017-08-30得票数 1
在Spark Structured Streaming (Java语言)中读取分区数据时，是否可以设置basePath选项？以下代码(使用常规的非流式数据帧)运行良好： .option("basePath", basePath)+---+---+---++---+---+---+
|
浏览 56提问于2018-03-01得票数 6
1回答
我正在尝试在Scala中运行一个Spark应用程序来连接到ActiveMQ。出于这个目的，我正在使用Bahir format("org.apache.bahir.sql.streaming.mqtt.MQTTStreamSourceProvider")。当我在我的built.sbt中使用Bahir2.2时，应用程序运行正常，但在将其更改为Bahir3.0或Bahir4.0时，应用程序无法启动，并且它给出一个错误：
浏览 24提问于2020-12-09得票数 1
回答已采纳
2回答
我使用Spark structured streaming处理从Kafka读取的记录。以下是我想要达到的目标： (i) Grou
浏览 2提问于2017-10-02得票数 15
2回答
Kinesis将文件的持久性(在本例中为时间序列JSON )管理为一个文件夹层次结构，该层次结构由YYYY/MM/DD/HH (直到24小时编号)...great划分。那么我如何使用Spark2.0来读取这些嵌套的子文件夹，并从所有的叶json文件中创建一个静态的Dataframe呢？数据阅读器有“选项”吗？我的下一个目标是成为一个流DF，火龙将新文件持久化到s3中，使用Spark2.0中</em
浏览 4提问于2016-10-30得票数 8
我有一个接收点击流的用例，我需要使用来自Cassandra的数据进行某些计算，最后将计算出的值推送到Redshift。 对于流和计算的这个用例，我正在探索所需的Tech Stack。可以使用Kafka Streams lib吗？ 如果有人用过这个，可以阐明可能的优点/缺点或任何其他建议。
浏览 14提问于2020-10-17得票数 1
我使用spark(3.0.0)结构化流从kafka读取主题。下面的火花官方指南部分没有提到DB sink，它也不支持为update mode：写到files附注：我跟踪了这些文章以获得aggregated流查询。stackoverflow
浏览 5提问于2020-09-16得票数 0
我有一个Spark Structured Streaming Job，它从S3读取数据，转换数据，然后将其存储到一个S3接收器和一个Elasticsearch接收器。当这样做时，似乎Spark从S3源读取数据两次，每个接收器一次。
有没有一种更有效的方法来写入同一管道中的多个接收器？
浏览 1提问于2017-09-19得票数 14
我想使用pyspark读取Azure Databricks中的流Twitter数据，但我不知道如何做到这一点？我在互联网上找到了几个链接，但它们都需要我的本地计算机的主机I或IP地址，但我想在云上做到这一点，有什么方法或其他代码可以做到吗？上面的https://towardsdatascience.com/sentiment-analysis-on-streaming-twitter-data-using-spark-structu
浏览 15提问于2021-07-18得票数 0
1回答
我是第一次接触kafka和spark structured streaming。我想知道spark in batch模式是如何知道从哪个偏移量读取的？如果我指定"startingOffsets“为”最早的“，我只会得到最新的记录，而不是分区中的所有记录。我在两个不同的集群中运行了相同的代码。集群A(本地机器)获取了6条记录，集群B( TST集群-第一次运行)获取了1条记录。，我会得到从昨天到当前运行
浏览 0提问于2020-10-24得票数 0
2回答
如何更新RDD？
、、
我们正在开发Spark框架，其中我们正在将历史数据移动到RDD集合中。现在有一个用例，其中RDD中的数据子集被更新，我们必须重新计算值。2.b.现在，我在步骤2.a中对这个新的RDD再次
浏览 45提问于2014-12-16得票数 20
我们目前使用Hazelcast ()作为分布式内存数据网格。这对我们来说工作得还不错，但在我们的用例中，仅在内存中运行已经走完了它的道路，我们正在考虑将我们的应用程序移植到NoSQL持久化存储。经过通常的比较和评估，我们即将选择Cassandra，并最终选择Spark进行分析。尽管如此，我们的架构需求中有一个缺口，我们仍然不知道如何在Cassandra中解决(无论有没有Spark)：Hazelcast允许我们创建一个连续的查询，因为无
浏览 3提问于2015-01-03得票数 7
1回答
我正在Azure数据库中尝试这段代码：df = spark.readStream.format("eventhubs在最后一步，我碰到了一个错误：
org.apache.
浏览 1提问于2020-05-13得票数 0
1回答
我正在使用Spark Structured Streaming来分析传感器数据，并且需要根据传感器以前的时间戳执行计算。我的传入数据流有三列: sensor_id、timestamp和temp。我需要添加第四列，即传感器以前的时间戳，这样我就可以计算每个传感器的数据点之间的时间。
这很容易使用传统的批处理，使用lag函数和按sensor_id分组。在流媒体的情况下，最好的方法是什么？("sensor_id&
浏览 2提问于2018-02-11得票数 8
我是Spark的新手，我有一个设置，我想要读入两个数据流，每个数据流来自Kafka主题，使用Spark structured streaming 2.4。val df1 = spark.readStream .option("kafka.bootstrap.servers", endpoint)
浏览 37提问于2021-03-21得票数 1
回答已采纳
我正在对来自kafka的数据运行结构化流式处理(spark 2.1.0 with zeppelin 0.7)，并尝试使用spark.sql可视化流式处理结果%spark2 .builder() .master("yarn&quo
浏览 4提问于2017-07-13得票数 6
云服务器
ICP备案
对象存储
实时音视频
云直播
腾讯云开发者
