Has anyone seen this below error while running vacuum ? java.util.concurrent.RejectedExecutionException in Spark

Timetravel/restore not possible after append Accidentally, we performed an append operation on a Delta Table that also added two columns to the target table:

df.write
  .format('delta')
  .partitionBy(partition_col)
  .mode('append')
  .option('userMetadata', json.dumps(commit_info))
  .option('mergeSchema', 'true')
  .option('overwriteSchema', 'true')
  .option("path", path)
  .saveAsTable(tablename)

Immediately, we realized our mistake and we tried "roll-back" the operation by restoring a previous version of the Delta Table. In the

_delta_log

there were JSONs for versions 745, 746, 747, 748, 749, 750 present at that time. We tried to restore version 749 (

deltaTable.restoreToVersion(749)

). The operation failed, saying that only version(s) [750, 750] are available. Also, reading a version prior to 750 did not work with the same error message. We did not perform any vacuums in the meantime. Does anyone have an explanation what the reason for this might be? I tried to recreate the situation with dummy data, but was not able to. In my experiments, I was always able to restore a previous version - even after a schema change.

We are moving 1tb data from 1 storage account to another (both mounted in Azure databricks) using dbutils.fs.mv(), this is slow operation. Meanwhile moving the same data manually by using "Microsoft Azure storage explorer" is 100x faster. Anyone has any suggestions for any APIs which can be used to perform this data movement using this higher speed transfer speed?

This message was deleted.

Is there anywhere to look into context of roadmap item: (TBD) Airbyte <> Delta Lake integration (@fvaleye) Interested in scope, progress and whether help is needed.

Hi Team, I am facing one weird issue, When am running MERGE sql inside Spark Structured Streaming foreachbatch it takes about 300+ seconds to complete. But when I ran it as independent sql in spark-shell it gets completes in just 25 seconds. Almost facing 10X degrade when am running inside my batch. But both cases am registering and using my batch df as physical table. Anyone faced like this issue before?.

Hi team, we are exploring delta-rs, is the delta-rs implementation have all the latest features of delta lake

I have lot of data file (~ in PB) in compressed csv format. The data file are organized by year/mm/dd/hh/mm/. I am trying to create an external table to query these data using below syntax and its taking a lot of time to execute. Any help to make it efficient ? The reason I am doing this is I want to create a table and share it with the management team so that they can query those csv using spark SQL. Thank you. CREATE EXTERNAL TABLE Test ( TIMESTAMP String, Message String, EventName String, Device String ) USING CSV OPTIONS('path'='/2023/')

hello Team , i have created a Spark job which send an alert on slack of deltaLake raw and refine streaming lag please let me know if anyone have better suggestion

hello Team , i am not able to see delta Table schema in glue only i can see array type col ,i am using EMR and running below query

%%sql
CREATE TABLE nm.delta_test (
  gender STRING,
  salary INT
) USING DELTA
LOCATION '<s3://nm-dev/nm/test>';

Please let me know if this possible to show the schema also in glue

Hi Team, I have a question. We are using delta lake open source in AWS ecosystem. We are using Athena 2.0 and to query delta lake tables we use symlink manifest files. I'm looking to share this table across to consumers. There are 2 types of consumers one who query from Athena, and another who consume the table to AWS EMR. Athena users need the symlink version and EMR users need the native delta table version without symlink. I dont want to create 2 different tables for both one poingint to the symlink and another pointing to the data files. Is there a way i can workaround this? Any suggestions would be really helpful

Hi Team, I have a question. I have a production setup where 4 different geo-distributed spark clusters each running a spark structured streaming application write into a single delta lake table. The table has partitions for each geo within which, there are partitions for year, month, date, hour and min. So the table layout looks like:

geo1/
 year1/
  month1/
   date1/
    hour1/
     min1
geo2/
geo3/
geo4/

In one of the applications(let's say for geo4), I'm observing an issue. I'm seeing way too many consecutive jobs titled: Delta: Compute snapshot for version: <snapshot version>. There are usually 20-30 such snapshot computation jobs each taking about a minute. This ends up making a single write essentially take about 30mins +- 10mins while the actual processing and saving within the job takes ~60s. The reason there are multiple snapshot computation jobs is because: while geo4 application is computing snapshot for version1, other applications write more versions(v2, v3, v4...). Now, geo4 has to catchup with them by computing snapshots for versions v2,v3,v4 in which time, other applications may write more new versions. Interestingly, the snapshot computation used to take much lesser earlier. I tried running a multi-cluster simulation where such snapshot computation jobs only take about a second. I'm curious why the SnapshotState computation takes so much time. Has anyone seen something similar earlier? My suscpicion is that the deltalog is now too big for my table with more than 330k entries.

Hey there, I’m facing an interesting problem where my end-users will often rewrite a partition with the same data multiple times per day. The problem is that the transaction log will get “flooded” with writes while having the same data in the table. End-users are using notebooks, one code cell writes the data and will get often executed because of rerunning the whole notebook. I wonder if there is a “smart” way to avoid writing data when it’s the same beside comparing the data row by row (which sound a bit complex)… Let me know if anyone has any ideas or tips, thank you!

Hi, Opened this issue for a spec question. https://github.com/delta-io/delta/issues/1727 #837 added multi-part checkpoint. Specification says that the checkpoint file can be split into multiple files. Are there any restrictions on how the actions must be split into different files? As an example, say we know that the only actions since last checkpoint have been add file actions. Then the additional datapoint in the new checkpoint is just the newly added files since last checkpoint. With multi-part checkpoint, can we copy the previous checkpoint as part 0 and add newly added files alone as part 1 of the checkpoint? If (1) is valid, are there ways to optimize this case, either now or in the roadmap? For example, a way to refer to the previous checkpoint file.

Hi all, I've changed the configs of my delta table (

minWriterVersion

,

minReaderVersion

,

columnMapping

) to support the

DROP COLUMN

command. It worked: when I query the table with Athena, the dropped column doesn't show up. But it still appears on Glue Catalog. If I try to run a Glue crawler on that table, it throws an error ("Internal Service Exception", no further information). I assume it's because Glue crawlers don't support the

DROP COLUMN

command and I need to revert the config changes. So my question is: is there any other non-manual way of updating the Glue catalog if I wanna use the `DROP COLUMN`command? Or perhaps some table property I can add on Glue catalog so that it works?

Gotta say, really looking forward to the pull request feat: delete operation for delta-rs. Thanks for the work @Bladja. With delete, we'll minimally be able to follow a delete-insert pattern for updates.

Hi Is it possible to create two delta table on same data ? Why is this required is, I want to keep the data in a shared storage, I want to create one delta table in my side and keep appending my data everyday at the storage through the delta table. At the same time allow other users to run queries on those data through maybe another delta table on their compute engine. Will it work if I create multiple delta table on the same data ?

Hello delta lake community of experts 🙂 I have a question: Are there any recommended best practies for dealing with large transaction logs for a delta lake table? Background: I have a large deltalake table in which multiple clusters are writing in a streaming manner. Thank you in advance for helping on this one.

Hi, we are getting with CDC the following error: "java.lang.IllegalArgumentException: CDC range from start 4 to end 1 was invalid. End cannot be before start." . We are reading from append only table by using the following code. We are not providing endingVersion:

spark.readStream.format("delta")

.option("readChangeFeed", "true")

.option("mergeSchema", "true")

.option("startingVersion", starting_version)

What could be the problem? Thanks for the help, Regards, Gapy

Hi, I am working on optimizing a Spark Delta Merge batch job using Delta Lake OSS 2.2 on EMR 6.10, and I am looking for some advice on how to tune the write performance. This is a daily batch job where 1 partition in our source table maps to multiple partitions in our target table. I wanted to start with questions on a few properties: 1. spark.databricks.delta.merge.repartitionBeforeWrite.enabled - should this be set to false? When set to true, I noticed it produced 1 large file per partition. 2. spark.databricks.delta.merge.materializeSource - I noticed the default for this was "auto". It seemed to consistently write the source data to disk, which caused much slower run times. I set this to "none" and then it seemed to read off memory. Any issues with this approach? 3. AQE - should we keep this on or off? With spark.databricks.delta.merge.repartitionBeforeWrite.enabled set to false, I noticed that the number of files produced per partition matched spark.sql.adaptive.coalescePartitions.initialPartitionNum (which is 1000 by default). 4. Any recommendations for how we tune spark.sql.shuffle.partitions? When spark.databricks.delta.merge.repartitionBeforeWrite.enabled = false, the value for spark.sql.shuffle.partition seems to map to number of files in each partition. 5. Are there any other properties in 2.2 we should explore that can have benefits on the Delta merge? 6. In our case, a large majority of the source data for the merge, goes to 5 partitions (for the current batch load date) in the target table. Is there any way we can leverage this information to optimize the merge process? @Justin Mark Santilli @Kimberly Mahoney @Shinoy Bhaskaran

Anyone know if there's any Delta Lake specific events happening in San Fran around the time of Data + AI convention?

Does Trino support delta sharing?

Does anyone has a reference code to implement distributed locking using zookeper for delta tables ?

Hi, for

delta-rs

when appending to a table it seems to go through every commit in history before writing (see thread for log). Is there a way to tell the update/append operation to start from a specific commit? Also, will this have performance impact if there are potentially tens of thousands of commits (assuming no debug logs are printed)?

Hello everyone, Is there a final version of the book “Delta Lake: The Definitive Guide by O’Reilly” or one with more than the 3 chapters of the early release? https://www.databricks.com/resources/ebook/delta-lake-the-definitive-guide-by-oreilly

Team, any alternative for this feature?. MERGE Cmd SQL will do?. https://github.com/delta-io/delta/issues/730

Can we control the number of row groups will be there in the checkpoint parquet files when they are written?

Hello! I stumbled upon some pyspark code where there is a merge to a delta table followed by an append to the same table, which appends the same data that is referred to in the insert clause in the merge. I was told that spark handles this in its execution plan, where all rows but one are appended, and that only one row will be inserted during merge - making the code run quicker as most rows are appended without further checks. Does this make sense? Does someone have an explanation for this? Thank you so much!

dt_target.alias("target")

.merge( source=df_stage.alias("updates"), condition=join_condition)

.whenMatchedUpdate( condition="...", set={...})

.whenNotMatchedInsert( values={...})

.execute()

df_load = df_stage.write.format("delta").mode("append") ...

Hello.. I tried to partition data based on Year and i see on S3 that there are folders named "Year=2017" etc (Please check below image) Is there any way to avoid the label like "Year" and get only "2017" as folder name ?

I have a partitioned delta table with about 10 million records, with the largest partition's parquet file coming in at ~170MB. This table gets rebuilt daily on a 15 node AWS glue cluster (spark 3.3.0, delta version 2.2.0) and takes about 10 minutes. Now I'm trying to merge intraday updates into this table using the same glue cluster. There are about 80k updated records being created, and I repartition them using the same partition column as the final delta table before attempting to merge. The problem is that the merge operation is hanging, and I can't figure out why. I've allowed the merge stage to run for 45 minutes before killing it. I've increased the driver memory to 20GB hoping that would solve the issue, but no luck. Has anybody seen this behavior, or have any suggestions on what I can try to make this merge work?