我正在对来自kafka的数据运行结构化流式处理(spark 2.1.0 with zeppelin 0.7)，并尝试使用spark.sql可视化流式处理结果%spark2.format("memory").queryName("test")运行后，我查询"test
浏览 4提问于2017-07-13得票数 6
1回答
需要在结构化流中对UDF中的表执行一些查询。问题是，在UDF内部，如果我尝试使用spark.sql，就会得到空指针异常。这里最好的方法是什么。基本上，我需要从一个表中流，然后使用这些数据从另一个表执行一些范围查询。val query = "select count(*) from appleMart where appleId='"+appleId+"'"
val
浏览 3提问于2020-07-29得票数 2
2回答
我有一个%sql块，我在用这段代码创建的数据框架上。df.createOrReplaceTempView("df")
我运行了许多查询来操作%sql块中的数据，我希望将该输出带到块之外并对其进行操作。具体来说，我想把它放到一个%r块中，并将数据可视化。我能让%sql块中的数据从块外部访问吗？
浏览 1提问于2019-08-30得票数 1
回答已采纳
1回答
火花流动态滤波器
、、、
我在下面的用例中使用火花流：
简单地说，假设我使用-其中"A“、"<”和"10“是动态的，来自某些数据库。实际上，这些值来自kafka主题，我正在使用并更新db中的</e
浏览 0提问于2018-07-10得票数 0
回答已采纳
1回答
我使用Spark流来处理在线需求，例如每小时新的用户计数如下：这种方法经常使用表格，花费太大sql2由于我不熟悉结构化流，因此结构化流不支持多重聚合，因此我使用如下方法：
readStream将<e
浏览 3提问于2017-09-08得票数 0
回答已采纳
我有一个结构化流数据，我想对每一行取一个列中的值(在本例中是一个时间戳，如1525670700)，并使用该值查询另一个静态数据，以获得与该值最近的时间戳，并将从该查询返回的一行合并到结构化流数据框架，例如weather_df.filter(weather_df.timestamp_unix > 1525670700).sort(col('timestamp_unix
浏览 1提问于2018-05-07得票数 1
回答已采纳
0回答
如何在Kafka stream上实现从单个作业执行多个流SQL查询。结构化流媒体是一种可靠的前进方式。例如，我在一个作业中对一个流运行10个查询。假设我只想运行9个查询，是否有一种方法可以动态地更改要在每次运行时从存储运行的查询。我希望每次执行流查询时，都从存储中动态挑选查询，即连续查询。
浏览 10提问于2016-12-20得票数 1
如何在同一个SparkSession中设置主机、密码和用户？如何将它们与CassandraConnector一起使用？我还试图创建一个目录，但是我找不到用session.execute而不是spark.sql来发出请求的方法。
有人能帮我吗？我使用Spark结构化流3.1.2并使用cassandra连接来丰富我的数据。
浏览 2提问于2021-07-09得票数 1
回答已采纳
一个关于星火结构流与HIVE表集成的查询。 csvDF.createOrReplaceTempView("updates")
val query= spark.sql，数据不会被插入到令人兴奋的目录(我的现有目录中有一
浏览 0提问于2018-12-28得票数 13
df_streaming \ .format("console") \
stream_writer.awaitTermination() 类似这样的df_streamingreturn feature_df 我知道agg(functions.max("weight"))可以得到它，但是我还想在cal_feature中使用其他的pandas函数。agg(max)也可以，结构化数据流支持是否适用？ spark-2.
浏览 8提问于2021-06-16得票数 0
我正在从Spark流迁移到结构化流，并且我面临以下代码的问题： .format("console")使用下面的AnalysisException错误
由: org.apache.spark.sql.AnalysisException:流源的查询必须使用writeStream.start(
浏览 3提问于2017-07-06得票数 2
回答已采纳
row_number() from Product
现在:与其使用解析函数，如_row_number()_或秩()-而且不使用关联子查询，是否有一种方法可以在标准sql中获得相同的结果？注意:对于如何模拟解析函数，有一个很好的问答：。然而，所有答案都使用关联子查询。其动机如下:我使用的是一个基于SQL的<
浏览 2提问于2019-04-20得票数 1
回答已采纳
我已经开始探索Spark结构化流，以编写一些在此之前一直使用的应用程序。Q1。对于结构化流应用程序中的每个接收器，它将独立地从一个源读取(例如。卡夫卡)。意思是说，如果你读到一个主题A，并写到三个地方(如ES，Kafka，S3)，它实际上将设置。
这会不会是一种性能下降？因为它需要3个独立的连接
浏览 1提问于2018-05-23得票数 6
回答已采纳
我需要从我的spark sql查询调用一个函数。我尝试过udf，但我不知道如何操作它。以下是场景： sum=effdate+trandate我的sparksql查询如下所示：
spark.sql("select sum(cm.effdate, cm.trandate)as totalsum, name from CMLEdG cm ...."
浏览 0提问于2016-10-04得票数 1
我一直在浏览spark 2.3.1关于结构化流的文档，但是无法找到有状态操作如何在内部与状态存储一起工作的详细信息。更具体地说，我想知道的是：(1)状态存储是否分布？似乎在以前版本的火花，它是每个工人，但暂时不知道。我知道它是由HDFS支持的，但是没有解释内存存储实际上是如何工作的。
实际上，它是内存中的分布式存储吗？我对去复制特别感兴趣，如果数据流来自于(比方说)一个大的数据集，
浏览 0提问于2018-08-17得票数 10
我需要将从下面的dataframe (假设列“时间戳”产生相同的值)的值存储到一个变量，稍后我需要在某个地方使用这个变量。TimestampType).as("timestamp"))它的刺的错误说：
org.apache.spark.sql.AnalysisException:必须使用writeStream.start()执
浏览 0提问于2019-01-08得票数 0
1回答
我正在使用Spark结构化流，并处理来自Kafka的消息。有一次，我的结果表如下所示，其中数据集中的每一行都有一个Spark查询。.|我需要执行这些查询中的每一个，并处理结果。但是，结构化流不允许将这些SQL收集到驱动端，而且我们不能在任何转换中打开一个新的SparkSe
浏览 2提问于2019-11-22得票数 1
回答已采纳
1回答
类似于Kafka的日志压缩，有很多用例要求它只保留对给定键的最后更新，并使用结果(例如，用于连接数据)。B | 2 | foobarA | 15 | foobeedoo----------------------------
B | 2
浏览 0提问于2018-06-12得票数 5
回答已采纳
我正在设法从Zeppelin注释中的Impala查询中生成输出表中的HTML链接(每行一个链接，它们将在表中自己的列中)。单击链接将导致相关的数据文件从文件系统下载到用户的计算机(这是容易的位)。棘手之处在于如何在Impala输出表中生成链接。是否有特定的SELECT查询可执行此操作？我可以为Zeppelin用
浏览 2提问于2021-03-05得票数 1
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
腾讯会议
云直播
对象存储活动推荐
腾讯云开发者
