有一个连续的数据流，在所有转换之后，它具有下一个模式： root |-- device_id: string (nullabletimestamp: string (nullable = true) 还有一组规则，即： if metric_id = 4077 and value > 10 and value < 25 这意味着如果流中的任何行满足该条件，则必须将该行推入不同的流中。如何识别符合警报条件(有<e
浏览 22提问于2020-12-23得票数 3
1回答
我希望使用python将流数据从套接字发送到Spark结构化流本地。当我用Netcat尝试这个的时候，它起了作用。但当我制作服务器套接字而不是Netcat时，它就不起作用了。spark版本是3.1.2I，在同一台机器上编写了两个不同的jupyter笔记本页。以下是我所做的和回应。推出星火结构流
在本stage.中建立了->
浏览 0提问于2021-10-13得票数 1
3回答
当我从Kafka主题创建一个流并打印它的内容时 os.environ['PYSPARK_SUBMIT_ARGS'] = '--packages org.apache.spark:spark-streaming-kafka-0-8_2.11:2.0.2 pyspark-shell' from pyspark.streamingimport Stream
浏览 2提问于2019-12-08得票数 1
回答已采纳
如何在DataFrame中为流PySpark设置架构。from pyspark.sql import SparkSessionfrom pyspark.sql.functions\ .getOrCreate()
# Create DataFrame r
浏览 3提问于2016-12-29得票数 3
回答已采纳
1回答
我正试图在本地jupyter笔记本中使用pyspark运行结构化流。代码从目录中读取csv文件，并将输出流写入控制台。True),StructField("age", IntegerType(), True),\当我尝试使用以下方法读取流时1).load("C:/
浏览 5提问于2022-07-25得票数 1
我正在使用pyspark读取Kafka的流数据，然后我想将这些数据汇到mongodb。在call File "/home/b3ds/hdp/spark/python/lib/pyspark.zip/pyspark/sql/utils.py"，第63行，在deco "/home/b3ds/hdp/spar
浏览 4提问于2018-06-04得票数 2
0回答
如何在Kafka stream上实现从单个作业执行多个流SQL查询。结构化流媒体是一种可靠的前进方式。例如，我在一个作业中对一个流运行10个查询。我希望每次执行流查询时，都从存储中动态挑选查询，即连续查询。
浏览 10提问于2016-12-20得票数 1
8回答
我想做多个聚合在星火结构化流。就像这样：
当我在结构化流中运行时，它会给出一个错误：“流DataFrames/Datasets不支持多个流聚合”。是否有一种方法可以在结构化流中实现这样的多个聚合？
浏览 14提问于2016-12-07得票数 21
1回答
我第一次使用火花放电。火花版本: 2.3.0卡夫卡版本: 2.2.0
我有一个卡夫卡生产者，它以avro格式发送嵌套数据，我正在尝试用火花流/结构化流写代码，这将反序列化从kafka到dataframe的avro，做转换，把它以拼花格式写成我能够在spark/scala中找到avro转换器，但是还没有增加对pyspark的支持。如何在火花放电中转换相同的内容。谢谢。
浏览 3提问于2019-05-02得票数 3
回答已采纳
1回答
from pyspark.sql import SparkSessionfrom pyspark.sql.functionslines = spark.readStream \ .table("myschema.streamTest") 在notebook 2中，我有在执行时，我也看不到notebook本身<e
浏览 21提问于2021-04-23得票数 1
回答已采纳
1回答
我有以下表格的火花数据格式：from pyspark.sql.functions import * sdf_cons.show()
I正在以下列形式接收结构化流数据：
浏览 2提问于2022-05-26得票数 0
我使用Spark 2.4.3和Kafka 2.3.0。我想用从Kafka到Spark的数据做Spark结构化流媒体。一般来说，它可以在测试模式下工作，但由于我必须对数据进行一些处理(并且不知道另一种方法)，Spark数据帧不再具有流式传输功能。 #!/usr/bin/env python3
from pyspark.sql.functions import from_jsondf2 = df1.wit
浏览 22提问于2019-09-10得票数 0
回答已采纳
我对PySpark结构化流有以下问题。有没有人知道如何做到这一点？我尝试使用结构化流文档的窗口函数示例，但是没有用。
非常感谢
浏览 0提问于2019-11-14得票数 1
org.apache.spark.sql.AnalysisException: Queries with streaming sources must be executed with writeStream.start();;它在Dataframes中运行良
浏览 0提问于2018-01-11得票数 3
回答已采纳
我有多个AWS kinesis数据流/firehose和CSV格式的结构化数据。我需要使用运动数据分析对这些数据执行分析。但是如何将多个流合并为一个流呢？因为Kinesis数据分析仅从一个流中获取数据。数据流可以存在于不同的地域。
问题:如何将Kinesis数据流合并为一个用于Kinesis数据分析的数据流？
浏览 3提问于2020-08-25得票数 0
我的目标是在某种数据结构中维护输入数据流，以便对它们执行查询和聚合操作。以连续数据流为输入，通过spark的结构化流，我将其存储在DataFrame中。我的问题是：
是一种易失性的数据结构吗?在程序崩溃的情况下，是维护的吗？是分布在集群各个节点上的DataFrame，还是保存在执行代码的节点上？可以在DataFrame上创建索引
浏览 1提问于2020-04-23得票数 0
1回答
类似于Kafka的日志压缩，有很多用例要求它只保留对给定键的最后更新，并使用结果(例如，用于连接数据)。B | 2 | foobarA | 15 | foobeedoo----------------------------
浏览 0提问于2018-06-12得票数 5
回答已采纳
我一直在开发Azure事件集线器的概念证明，使用Pyspark将json数据流到Azure Databricks笔记本。在我看到的示例中，我创建了我的粗略代码，如下所示，将数据从事件集线器接收到我将用作目的地的delta表ehConf = {'eventhubs.connectionString("append") \
.option("checkpointLocation&
浏览 1提问于2019-11-12得票数 1
回答已采纳
在文档中，它说像结构化流中的mapGroupsWithState这样的有状态操作只支持Scala和Java，但我确实需要python中的有状态功能。我该怎么办？
浏览 0提问于2018-04-12得票数 4
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
对象存储
即时通信 IM活动推荐
腾讯云开发者
