有没有可能有两个Spark进程同时读取一个Delta Table中的流？
在Delta Lake中，一个Delta表可以同时由多个Spark进程读取流数据。这是因为Delta Lake提供了ACID事务支持和并发控制机制，确保数据的一致性和并发读取的正确性。
具体来说，Delta Lake使用了一种称为“读时快照隔离（Read Committed Snapshot Isolation，RCSI）”的隔离级别。在RCSI下，每个读取操作都会创建一个快照，该快照反映了读取操作开始时的表状态。这意味着，即使其他进程在读取过程中修改了表的数据，读取操作也不会受到影响，因为它们读取的是快照而不是实时数据。
另外，Delta Lake还使用了一种称为“乐观并发控制（Optimistic Concurrency Control，OCC）”的机制来处理并发写入。当多个进程同时尝试写入Delta表时，Delta Lake会检测到冲突并拒绝其中一些写入操作，以确保数据的一致性。
因此，可以放心地使用两个Spark进程同时读取一个Delta Table中的流数据，Delta Lake会保证数据的一致性和并发读取的正确性。
关于Delta Lake的更多信息和推荐的腾讯云相关产品，您可以参考以下链接：
- Delta Lake概述：https://cloud.tencent.com/document/product/1270/47247
- 腾讯云数据湖分析 Delta Lake：https://cloud.tencent.com/product/datalake
- 腾讯云数据仓库 TDSQL-C：https://cloud.tencent.com/product/tdsqlc
相关·内容
我正在接收一个增量表中的数据，并且我希望有两个消费者处理该数据的读取： 我把这些代码放在不同的罐子里： 1)实时计算聚合的Spark进程。 val df_aggregations = spark.readStream
.format("delta")
.option("ignoreDeletes", "true")
.option("ignoreChanges", "true")
.option("checkpointLocation", configura
浏览 27提问于2019-04-18得票数 1
回答已采纳
我有两个火花流设置在笔记本上运行类似这样的并行。
spark.sparkContext.setLocalProperty("spark.scheduler.pool", "pool1")
df1 = spark \
.readStream.format("delta") \
.table("test_db.table1") \
.select('foo', 'bar')
writer_df1 = df1.writeStream.option(
浏览 3提问于2019-10-07得票数 1
回答已采纳
我有一个用例，我们需要将Open Source Delta表流式传输到多个查询中，并对其中一个分区列进行过滤。例如，.给定的增量表在年份列上分区。
Streaming query 1
spark.readStream.format("delta").load("/tmp/delta-table/").
where("year= 2013")
Streaming query 2
spark.readStream.format("delta").load("/tmp/delta-table/").
where(
浏览 0提问于2021-02-24得票数 1
1回答
据我所知，Delta体系结构中的铜表以表格格式表示原始数据和(或多或少)未经修改的数据。这是否意味着我也不应该对铜表的数据进行分区？您可以将分区看作依赖于用例的东西，它指向Silver甚至Gold表。
请看下面的示例：
def read():
return spark.read\
.format("csv")\
.option("delimiter", "\t")\
.option("header", True)\
.load("file.tsv.gz
浏览 4提问于2020-12-14得票数 1
回答已采纳
在火花流作业中，有什么方法可以在到Delta表之后得到更新/插入行？
val df = spark.readStream(...)
val deltaTable = DeltaTable.forName("...")
def upsertToDelta(events: DataFrame, batchId: Long) {
deltaTable.as("table")
.merge(
events.as("event"),
"event.entityId == table.entityId"
浏览 3提问于2021-09-19得票数 2
回答已采纳
我目前正在探索delta，这是由databricks开源的。我正在读取kafka数据，并使用delta lake格式将其写入为流。Delta lake在从kafka进行流式写入的过程中创建了许多文件，我觉得kafka是hdfs文件系统的核心。 我已经尝试过将多个文件压缩为单个文件。 val spark = SparkSession.builder
.master("local")
.appName("spark session example")
.getOrCreate()
val df = spark.read.parque
浏览 16提问于2019-10-13得票数 4
回答已采纳
我有一个三角表，我正在阅读作为StreamingQuery。
纵观Delta历史，使用DESCRIBE History，我看到99%的OperationMetrics声明numTargetRowsUpdates is 0 (大多数操作都是插入的)。然而，有时会出现2-3个numTargetRowsUpdates > 1的情况，而Delta上的操作则是一个合并。
我仍然可以使用StreamingQuery并将这些数据作为流读取，还是会得到错误？即：
df: DataFrame = spark \
.readStream \
.
浏览 6提问于2022-04-14得票数 3
我使用并设置选项maxBytesPerTrigger：temp_data = spark.readStream.format("delta").option("maxBytesPerTrigger",1000).load(raw_data_delta_table)，但是我的整个文件都是在一个批处理中加载的。我想把它装进多个批次。我错过了什么地方？是的，如果我使用maxFilesPerTrigger，它就能正常工作。但是maxBytesPerTrigger不起作用。谢谢
浏览 0提问于2021-05-26得票数 2
1回答
有没有办法只读取两个版本的deltaLake之间的增量？尤其是在Synapse中
我试过了
%%pyspark
import delta
from pyspark.sql.functions import col, asc
df_delta = spark.read.format("delta") \
.option("readChangeFeed", "true") \
.option("startingVersion", 0) \
.option("endingVersion", 1) \
.
浏览 0提问于2021-05-24得票数 1
1回答
我有一个三角表名为"ali“。我从三角表中读到了流：
from pyspark.sql.functions import col
streamDF = spark.readStream.format('delta').load('dbfs:/user/hive/warehouse/pdf/ali')
display(streamDF)
现在，我希望将数据流写入我的银delta表中：
(streamDF
.writeStream
.format("delta")
.option("checkpointLocation" ,
浏览 8提问于2022-09-20得票数 0
回答已采纳
我有一个简单的星火作业，它将数据流到Delta表。这张表很小，没有分区。
创建了许多小的拼花文件。
按照文档()中的建议，我添加了一个压缩作业，每天运行一次。
val path = "..."
val numFiles = 16
spark.read
.format("delta")
.load(path)
.repartition(numFiles)
.write
.option("dataChange", "false")
.form
浏览 5提问于2021-08-12得票数 4
2回答
创建1GB分区Spark
、、、
在用S3编写spark时，我试图将数据拆分为1GB。我尝试的方法是在GB ( define_coalesce函数)中计算define_coalesce的大小，然后使用这个数字用S3编写：
# Vaccum to leave 1 week of history
deltaTable = DeltaTable.forPath(spark, f"s3a://{delta_table}")
deltaTable.vacuum(168)
deltaTable.generate("symlink_format_manifest")
# Reading delta t
浏览 8提问于2022-09-20得票数 0
我使用的是delta lake oss版本0.8.0。 让我们假设我们使用原始数据计算了聚合数据和立方体，并使用delta将结果保存在一个金表中。 我的问题是，有没有一种众所周知的方法来访问这些金表数据并将其传送到web仪表板？ 在我看来，您需要一个正在运行的spark会话来查询增量表。 因此，一种可能的解决方案是编写一个web api，它执行这些spark查询。 您也可以将gold结果写入postgres之类的数据库中以访问它，但这似乎只是复制数据。 是否有已知的最佳实践解决方案？
浏览 31提问于2021-02-24得票数 1
回答已采纳
我有一个从Azure Eventhubs到ADLS的流式查询流数据，每5秒一次，相同的流查询是1小时窗口的水印，有5分钟的水印延迟。
代码：
val rawStreamQuery = messages.writeStream.format("delta")
.option("checkpointLocation", BASE_LOC + "checkpoint/" + RAW_SCHEMA_NAME + "/" + RAW_TASK_TABLE)
.trigger(Trigger.ProcessingTim
浏览 2提问于2021-05-17得票数 1
1回答
我们让传感器在一天内多次随机启动和运行。来自传感器的数据被发送到Kafka主题，由Spark Structured streaming API使用，并存储到Delta Lake。现在，我们必须确定每个传感器的会话，并将其存储在由device_id和sensor_id分区的不同Delta Lake表中。
我尝试了Spark Structured加水印，但效果不是很好。
stream2 = spark.readStream.format('delta')
.load('<FIRST_DELTA_LAKE_TABLE>')
浏览 6提问于2021-02-17得票数 0
1回答
我创建该表的方式如下：
spark.sql("CREATE TABLE IF NOT EXISTS table USING DELTA AS SELECT * FROM origin")
但我知道这个错误：
线程"main“org.apache.spark.SparkException中的异常:表实现不支持写:表
浏览 4提问于2021-08-30得票数 0
回答已采纳
2回答
我创建了一个增量表，它使用spark 3.x和增量0.7.x创建：
data = spark.range(0, 5)
data.write.format("delta").mode("overwrite").save("tmp/delta-table")
# add some more files
data = spark.range(20, 100)
data.write.format("delta").mode("append").save("tmp/delta-table")
df = sp
浏览 1提问于2020-08-28得票数 0
回答已采纳
1回答
如何在中的Delta上运行S3命令？当我这么做
deltatable = DeltaTable.forPath(spark, "s3a://test-bucket/delta_table/")
spark.catalog.refreshTable(deltatable)
我发现了一个错误：
AttributeError: 'DeltaTable' object has no attribute '_get_object_id'
refresh命令只适用于Hive表吗？
谢谢!
浏览 3提问于2022-08-09得票数 1
回答已采纳
我试着把"small_radio_json.json“装到三角湖桌上。在这段代码之后，我将创建表。
我尝试创建Delta表，但得到了错误“写入Delta表时检测到的架构不匹配”。它可能与events.write.format("delta").mode("overwrite").partitionBy("artist").save("/delta/events/")的分区有关。
如何修复或修改代码。
//https://learn.microsoft.com/en-us/azure/azure-databricks
浏览 1提问于2020-03-29得票数 9
回答已采纳
1回答
在我的一个用例中，我使用的是delta lake的change ( CDF )特性，它很适合使用CDF，但是当我读取所有要插入的数据时，它列出了所有版本，是否有一种方法可以只读取最新版本而不指定版本号或获取最新版本？
return spark.read.format("delta") \
.option("readChangeFeed", "true") \
.table(tableName) \
.where(col(&#
浏览 7提问于2022-06-05得票数 0
将数据保存为delta格式时遇到错误：
$ data.write.format("delta").save("gs://xxx/delta-table1")
org.apache.spark.SparkException: Job aborted.
at org.apache.spark.sql.execution.datasources.FileFormatWriter$.write(FileFormatWriter.scala:198)
...
Caused by: java.lang.ArrayStoreException: org.apache
浏览 0提问于2020-11-13得票数 2
1回答
我需要在python中实时插入数据(使用火花结构化流)--这些数据是实时读取的(格式为csv)，然后编写为一个增量表(这里我们想更新数据，这就是为什么我们使用merge )，我使用的是增量引擎和databricks，我编写了如下代码：
from delta.tables import *
spark = SparkSession.builder \
.config("spark.sql.streaming.schemaInference", "true")\
.appName("SparkTest") \
.getOrCreate()
浏览 5提问于2021-12-10得票数 4
回答已采纳
在研究之后，我注意到使用java的ProcessBuilder的“正确”方法是生成另外两个线程来管理新创建的进程的stdout/stderr，这样它就不会挂起，如下所示：
但这让我想知道两个问题--一个。)为什么需要单独的进程，而不是让父进程抢占stdout，然后依次使用stderr呢？
2.)此外，如果要将流重定向到stdout，那么只让父进程吞下stdout流，而不必担心死锁是否可以接受？
浏览 3提问于2010-06-16得票数 3
我正在研究将Apache Spark用于一个应用程序。我对使用临时视图和完整SQL查询的结构化流模式特别感兴趣(为了简单和低延迟)。 应用程序将需要在单个输入数据流上运行多个(数十个，可能数百个)查询。有没有办法避免Spark重复读取每个查询的输入？
浏览 27提问于2021-01-20得票数 0
回答已采纳
我一直在开发Azure事件集线器的概念证明，使用Pyspark将json数据流到Azure Databricks笔记本。在我看到的示例中，我创建了我的粗略代码，如下所示，将数据从事件集线器接收到我将用作目的地的delta表
connectionString = "My End Point"
ehConf = {'eventhubs.connectionString' : connectionString}
df = spark \
.readStream \
.format("eventhubs") \
.options(**ehC
浏览 1提问于2019-11-12得票数 1
回答已采纳
我希望使用类似线程中发布的这个语法来更新我所拥有的delta表的分区。挑战是databricks中不存在delta表，但是我们使用databricks来更新可以通过Azure数据工厂更新的增量。
如何调整下面的语法以更新分区并通过表路径覆盖表？
Python：
val input = spark.read.table("mytable")
input.write.format("delta")
.mode("overwrite")
.option("overwriteSchema", "true")
浏览 6提问于2022-12-01得票数 1
1回答
我在s3桶上有一个三角洲湖。因为我想使用Spark的SQL，所以我需要同步达美湖和本地星火会话。有没有一种快速的方法可以让所有的表都可用，而不必为每个表创建一个临时视图？
目前，这就是我所做的(假设我在s3_bucket_path“文件夹”中有3个表)。
s3_bucket_path = 's3a://bucket_name/delta_lake/'
spark.read.format('delta').load(s3_bucket_path + 'table_1').createOrReplaceTempView('table_1&#
浏览 5提问于2022-10-13得票数 0
回答已采纳
使用Databricks Delta table，您可以使用合并操作将源表、视图或DataFrame中的数据向上插入到目标增量表中。此操作类似于SQL MERGE INTO命令，但对更新、插入和删除中的删除和额外条件有额外的支持。
我可以使用以下Python代码成功执行合并：
from delta.tables import *
deltaTable = DeltaTable.forPath(spark, delta_path)
(deltaTable
.alias("t")
.merge(loanUpdates.alias("s"), "t
浏览 3提问于2020-08-19得票数 0
我们是否需要在Spark Structured Streaming中对Kafka的readStream和writeStream都设置检查点？何时需要对这两个流都设置检查点，还是只对其中一个流设置检查点？
浏览 2提问于2017-06-16得票数 3
回答已采纳
我是用前置批处理处理增量表中的流的。
spark.readStream.format("delta") \
.option("readChangeFeed", "true").option("startingVersion","latest") \
.load("dbfs:/mnt/cg/file_to_up/table2") \
.writeStream.queryName("myquery") \
.foreachBatch(merge_deltas) \
浏览 3提问于2022-04-20得票数 2
这是我在笔记本中使用的电火花代码
data_lake_container = 'abfss://abc.dfs.core.windows.net'
stage_folder = 'abc'
delta_lake_folder = 'abc'
source_folder = 'abc'
source_wildcard = 'abc.parquet'
key_column = 'Id'
key_column1 = 'LastModifi
浏览 17提问于2022-07-12得票数 0
2回答
如何找出达美表的最后两个版本之间的区别？以下是我使用dataframes时所使用的内容：
val df1 = spark.read
.format("delta")
.option("versionAsOf", "0001")
.load("/path/to/my/table")
val df2 = spark.read
.format("delta")
.option("versionAsOf", "0002")
.load("/path/to/m
浏览 1提问于2021-11-17得票数 5
回答已采纳
我们有一个增量活动表，它从kafka主题读取、清理/过滤/处理/聚合消息，并将其转储到铜/银/金表，为了构建REST服务来检索聚合结果，我们需要将数据从黄金表转储到cassandra表。我试图更新gold表的脚本，在将聚合结果转储为gold之后，我又添加了一个步骤来进一步将更新的结果转储到cassandra表，但它没有工作：
@dlt.table
def test_live_gold():
return (
dlt.read("test_kafka_silver").groupBy("user_id", "event_type").c
浏览 1提问于2022-09-07得票数 1
回答已采纳
1回答
我正在尝试通过spark java将增量表恢复到以前的版本，我使用的是本地ide .code，如下所示 import io.delta.tables.*;
DeltaTable deltaTable = DeltaTable.forPath(spark, <path-to-table>);
DeltaTable deltaTable = DeltaTable.forName(spark, <table-name>);
deltaTable.restoreToVersion(0) // restore table to oldest version
del
浏览 43提问于2021-04-27得票数 2
2回答
我在readStream的增量表上遇到了问题。 什么是预期的，参考来自以下链接https://docs.databricks.com/delta/delta-streaming.html#delta-table-as-a-stream-source Ex： spark.readStream.format("delta").table("events") -- As expected, should work fine 问题，我用以下方式尝试了相同的方法： df.write.format("delta").saveAsTable("de
浏览 34提问于2020-06-12得票数 5
回答已采纳
1回答
在spark项目中使用kafka和delta_core依赖项时，我收到了下一个警告： [WARNING] delta-core_2.12-0.7.0.jar, spark-sql-kafka-0-10_2.12-3.1.1.jar define 1 overlapping resources:
[WARNING] - META-INF/services/org.apache.spark.sql.sources.DataSourceRegister 这会导致不能找到增量源。我如何才能同时包含delta和kafka？谢谢。 下面是我的maven配置： <depende
浏览 48提问于2021-04-26得票数 0
我在HDFS上以Delta格式写入了数据。据我所知，Delta正在以拼接的形式存储数据，只是在它上面增加了一个具有高级功能的层。 但是，当使用Pyspark读取数据时，如果使用spark.read.parquet()或spark.read.format('delta').load()读取dataframe，我会得到不同的结果 df = spark.read.format('delta').load("my_data")
df.count()
> 184511389
df = spark.read.parquet("my_data
浏览 28提问于2021-01-27得票数 1
回答已采纳
我们每天在delta lake表(表名: dst1)中加载源文件作为时间序列数据。如果deltalake表不存在，我们将创建一个表。如果表确实存在，我们将合并新数据作为时间序列数据。 Creating Table if table does not exist:
spark.sql(f"CREATE TABLE {dtable} USING DELTA LOCATION {dmount1}")
If table does exist we will merge the new data as time series data:
here df -->
浏览 20提问于2021-09-23得票数 2
回答已采纳
3回答
我有两个流数据集，我们称它们为fastStream和slowStream。
fastStream是一个流数据集，我是通过结构化流API从卡夫卡消费的。我希望每秒钟可能会收到数千条信息。
slowStream实际上是一个引用(或查找)表，它被另一个流“向上插入”，包含在将记录保存到表之前要连接到fastStream中的每个消息的数据。只有当有人更改元数据时，slowStream才会被更新，元数据可以在任何时候发生，但我们希望每隔几天更改一次。
fastStream中的每条记录都将在slowStream中有一个对应的消息，我基本上希望在slowStream表中的任何数据中立即实现该连接。我不想等着
浏览 6提问于2021-03-30得票数 3
回答已采纳
我无法在Databricks笔记本中通过pySpark读取URL的内容(Version8.3，Spark3.1.1)。我已经尝试了几乎所有的可能性，但无法找到确切的问题。这是我的密码。
from pyspark import SparkFiles
url = 'https://pds-atmospheres.nmsu.edu/PDS/data/mors_1101/tps/1998_028/8028d38a.tps'
spark.sparkContext.addFile(url)
df1 = spark.read.text("file://"+SparkFiles
浏览 9提问于2021-10-20得票数 0
我正在使用python api和pyspark进行增量湖合并操作。在执行合并操作之后，我调用了压缩操作，但压缩操作给出了以下错误：
错误：
File "/usr/lib/spark/python/lib/pyspark.zip/pyspark/sql/readwriter.py", line 170, in load
File "/usr/lib/spark/python/lib/py4j-0.10.7-src.zip/py4j/java_gateway.py", line 1248, in __call__
File "/usr/lib/s
浏览 3提问于2020-06-10得票数 0
我开始查看delta，并在尝试更新表时得到此异常。 我使用的是： aws EMR 5.29 Spark 2.4.4 Scala版本2.11.12，使用io.delta:delta-core_2.11:0.5.0。 import io.delta.tables._
import org.apache.spark.sql.functions._
import spark.implicits._
val deltaTable = DeltaTable.forPath(spark, "s3://path/")
deltaTable.update(col("col1"
浏览 22提问于2020-01-21得票数 0
1回答
如何使用UDF处理大型增量表？
、、、、
我有一个约3,000亿行的delta表。现在，我正在使用UDF对一列执行一些操作，并创建另一列。
我的代码是这样的
def my_udf(data):
return pass
udf_func = udf(my_udf, StringType())
data = spark.sql("""SELECT * FROM large_table """)
data = data.withColumn('new_column', udf_func(data.value))
现在的问题是，这需要很长时间，因为Sp
浏览 4提问于2022-03-24得票数 2
回答已采纳
我有一个关于delta lake的hive metastore支持的问题，我在一个独立的spark会话上定义了一个metastore，配置如下 pyspark --conf "spark.sql.catalog.spark_catalog=org.apache.spark.sql.delta.catalog.DeltaCatalog" --conf "spark.sql.extensions=io.delta.sql.DeltaSparkSessionExtension"
spark = SparkSession \
.builder \
.a
浏览 36提问于2020-09-06得票数 0
回答已采纳
我特别希望通过更新数据并将数据插入到具有大约4万亿条记录的DeltaLake基表中来优化性能。
环境: Spark 3.0.0 DeltaLake 0.7.0
在上下文中，这是关于通过DeltaLake创建一个增量表的，我将以步骤总结这一点，以便更详细地说明：
创建基表(增量)
获取周期数据
将数据添加到基表中
步骤1和步骤2已经完成，但是当添加数据时性能是众所周知的慢，例如添加一个9GB的csv大约需要6个小时，这主要是因为delta需要为每个更新重写数据，它还需要从数据库“读取”所有数据。
此表还被分区(由)并存储在集群的GDFS (HDFS)中，以确保火花节点能够执行
浏览 1提问于2020-09-11得票数 1
这里是星火2.4.4和德尔塔湖0.5.0。
我正在尝试使用delta数据源创建一个表，而且似乎遗漏了一些东西。尽管CREATE TABLE USING delta命令运行良好，但既没有创建表目录，也没有insertInto工作。
下面的CREATE TABLE USING delta运行良好，但insertInto失败了。
scala> sql("""
create table t5
USING delta
LOCATION '/tmp/delta'
""").show
scala> spark.catalog.li
浏览 4提问于2019-12-31得票数 11
回答已采纳
1回答
Spark版本: 3.2.1 Delta版本: 1.2.1 (也尝试了2.0版本)
当我试图运行入门代码来尝试“增量”时。
from pyspark.sql import SparkSession
from delta import *
builder = SparkSession.builder.appName("MyApp") \
.config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
.config("spark.sql
浏览 4提问于2022-06-29得票数 2
我使用的是增量OSS(v2.0.0)，我有一个现有的增量表，我希望为该表启用启用更改数据提要 (CDF)。但是，在更改表属性之后，我可以看到表属性已经更新，但是delta表的历史记录没有显示启用了CDF。
代码：
spark.sql("DESCRIBE HISTORY '/Users/yatharthmaheshwari/data-partner-merge/src/main/resources/delta/onaudience/dpm/base' ").show(2,false)
spark.sql("CREATE TABLE default.dp
浏览 11提问于2022-08-04得票数 1
回答已采纳
我目前是一名初级数据开发人员，最近看到一篇文章说Azure Synapse现在可以从Delta表创建SQL表。我尝试从Delta lake Storage V2中的delta表创建一个SQL表，但是当使用'PARQUET‘作为文件格式并使用通配符读取文件时，该表被填充了额外的冗余数据(来自文件夹中所有快照的所有数据)。
我尝试为我的表创建一个外部文件格式，但是Synapse不接受'DELTA‘作为数据类型。我使用了'PARQUET‘作为文件格式，并在我的Delta表上使用了VACUUM，只保存了它的最新快照。只要我设置了特定文件的路径，或者Delta表中只有一个snap
浏览 3提问于2021-02-26得票数 3
1回答
在中间过程粒度上的火花流
、、、
我在我的流浪开发机器上有两个核心，并且想运行两个流应用程序。
如果：
它们都采用了两个可用的核心(我没有指定"spark.cores.max")
它们的播放间隔为15秒。
5秒足够执行计算。
Mesos的预期行为是否会在两个应用程序之间转移这两个可用的内核？我期望这种行为，因为"Mesos锁定资源直到作业被执行“，而在星火流中，一个作业是在批处理间隔内执行的。
否则，如果资源在应用程序的生命周期中被锁定(在星火流中是永远的)，那么使用Mesos而不是独立的集群管理器有什么好处呢？
浏览 2提问于2016-05-05得票数 1
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
腾讯会议
实时音视频活动推荐
腾讯云开发者
