我正在接收一个增量表中的数据，并且我希望有两个消费者处理该数据的读取： 我把这些代码放在不同的罐子里： 1)实时计算聚合的Spark进程。val df_aggregations = spark.readStream .option("ignoreDeletes", "true&q
浏览 27提问于2019-04-18得票数 1
回答已采纳
在火花流作业中，有什么方法可以在到Delta表之后得到更新/插入行？val df = spark.readStream(...) .merge(
浏览 3提问于2021-09-19得票数 2
回答已采纳
2回答
我有下面的活桌我想把它写进一条流里，然后写回我的卡夫卡源。我在apache文档中看到，我可以使用writeStream (我已经使用readStream将它从我的kafka流中提取出来)。但是，我如何将表转换成它所需要的介质，以便它可以使用它呢？我对卡夫卡和数据世界都很陌生，所以这里欢迎任何进一步的解释。流中提取出来)。但是，我如何将表转换成它所需要的介质，以便它能够使用它呢?我对kafka和数据世界都很陌生，所以这里欢迎任何进一步的解释。
浏览 11提问于2022-11-03得票数 1
回答已采纳
我正在使用databricks spark 3.x，并且我正在读取大量的流(100+)，并且每个流都有自己的契约，并且需要写到它自己的增量/parquet/sql/任何表中。虽然这是很多流，但每个流的活跃度很低-有些流一天可能只看到数百条记录。我确实想要流，因为我的目标是一种相当低延迟的方法。(&q
浏览 25提问于2020-10-25得票数 1
1回答
流和静态数据集的合并是结构化流的一个很好的特点。但是每一批数据集都会从数据源中刷新。由于这些源并不总是动态的，因此在指定的时间段(或批数)缓存静态数据集将是一种性能增益。在指定的批处理期间/批数之后，数据集将从源重新加载，否则将从缓存中检索。
在星火流中，我使用缓存的数据集来管理它，并在指定数量的批处理运行后取消它<e
浏览 1提问于2017-12-13得票数 9
情况:我正在生成一个增量文件夹，其中包含来自以前的流查询A的数据，稍后从另一个DF读取数据，如下所示1 --当我试图以这种方式从相同的程序读
浏览 2提问于2019-12-28得票数 0
1回答
我正在尝试Databricks开发人员基金会Capstone，但我似乎无法通过流练习。df = (spark.readStream.schema(DDLSchema).option("maxFilesPerTrigger", 1).json(stream_path对它进行了转换
ordersQuery = (orders_df.write
浏览 1提问于2021-10-06得票数 0
我有一个用例，我们需要将Open Source Delta表流式传输到多个查询中，并对其中一个分区列进行过滤。例如，.给定的增量表在年份列上分区。Streaming query 1where("year= 2013")
St
浏览 0提问于2021-02-24得票数 1
我有一个简单的星火作业，它将数据流到Delta表。这张表很小，没有分区。按照文档()中的建议，我添加了一个压缩作业，每天运行一次。val numFiles = 16 spark.read .load(path)每次压缩作业运行时，流
浏览 5提问于2021-08-12得票数 4
我有一个非常具体的问题，我试图同时读取两个流，一个从管道进程读取到stdin，另一个从Popen启动的子进程读取。将该程序命名为“stream_compare.py”while True:
line = sys.stdin.readline().，在启动程序时
浏览 19提问于2018-01-11得票数 1
回答已采纳
1回答
Databricks有结果缓存的概念吗？当我运行一个SQL查询时，它是在某个地方缓存结果集以便进行子秒访问，还是只有Delta湖缓存？我在文档中找不到任何东西，在这个阶段，我假设它不作为一个特性存在。
浏览 4提问于2019-10-17得票数 1
1回答
我需要在python中实时插入数据(使用火花结构化流)--这些数据是实时读取的(格式为csv)，然后编写为一个增量表(这里我们想更新数据，这就是为什么我们使用merge )，我使用的是增量引擎和databricks，检查点路径按预期填写，delta表中的显示也会给出结果。display(table("deltaTable"))
在星星之火UI中，我看到写的步骤：
浏览 5提问于2021-12-10得票数 4
回答已采纳
1回答
在我的一个用例中，我使用的是delta lake的change ( CDF )特性，它很适合使用CDF，但是当我读取所有要插入的数据时，它列出了所有版本，是否有一种方法可以只读取最新版本而不指定版本号或获取最新版本return spark.read.format("delta") \
.option("readChangeFeed&q
浏览 7提问于2022-06-05得票数 0
1回答
我是Apache Spark Structured Streaming的新手，我正在做一个基本的POC。我的需求是构建一个规则引擎，它将从HDFS读取数据，HDFS接收来自多个源的连续数据流。我需要根据此HDFS数据的可配置时间窗口运行不同的规则。例如，可能有一个规则需要在最后5分钟的数据上运行，而第二个规则将在最后7分钟的数据
浏览 31提问于2021-09-16得票数 1
3回答
是否有任何方法将新列附加到现有的拼花文件中？我知道Spark随而来，但是这个例子只给出了一个键值的情况。
拼花“附加”模式也做不到这一点。它只将新行附加到拼花文件中</
浏览 1提问于2015-08-04得票数 14
我目前正在探索delta，这是由databricks开源的。我正在读取kafka数据，并使用delta lake格式将其写入为流。Delta lake在从kafka进行流式写入的过程中创建了许多文件，我觉得kafka是hdfs文件系统的核心。 我已经尝试过将多个文件压缩为单个文件。()
val df = spark.read.parquet("deltalakefile/data/
浏览 16提问于2019-10-13得票数 4
回答已采纳
我运行的是SPAR3.1.1和AWS emr-6.3.0集群，具有以下蜂巢/亚稳态配置：s
浏览 3提问于2022-02-09得票数 2
回答已采纳
我正在本地模式下在一个ec2实例上运行Spark3.0。今天，我希望能够在并行中运行多个python脚本，以便提供对我使用的spark会话的访问- .config("spark.jars.packages", "io.delta:delta-core_2.12:0.7.0"
浏览 2提问于2020-09-19得票数 0
回答已采纳
1回答
有没有办法可以使用表名而不是路径来读取增量表的版本？我寻找这个用例的原因是，分析师团队只想知道2021-02-07的表的完整快照，但他/她只知道表名，而他们不知道实际数据所在的ADLS路径。我们可以从Databricks文档中读取如下版本：df = spark.read.format("delta").option("t
浏览 2提问于2021-04-09得票数 1
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
云直播
对象存储
腾讯会议活动推荐
腾讯云开发者
