There has been not much help in this group . No one responds ! 😞

@Parthiban Jaganathan Thanks for the hint !

Hi Team, If we set overwriteSchema=true and we change the datatype of any column then does it rewrite the all files to change the schema of the table?

HI I have 2 questions. 1. In my project, I need to bring data for MLOps with Delta Lake. I have JSON-formatted event data and video data from blob storage. To manage the JSON data, I'm considering storing each event JSON data within a single cell of Parquet based on ID. Then, I can load and analyze it using Pandas when needed. I need updating the event JSON within the Parquet cell if there are any updates. Please let me know your thoughts on my idea. Is it reasonable? 2. I want to know if there's any method to update parquet with delta-rs(without using Spark, for sure). So far, I have been using pandas to create and merge the data before overwriting for update. However, due to the large volume of event data being generated, it seems impractical to handle it that way.

Hi all Suppose we need to modify a Delta Table - column "A" from Decimal to Integer. We want to ensure a smooth transition without causing disruptions to both the writers and readers in our production environment, especially considering the large volume of data involved (billions of data rows). For contributors (writers), we have the ability to pause the streaming ingestion, so currently we se no issue in that regard. For consumers (readers), we would love to minimize interruptions as much as possible. What would the best practice be ? (Databricks)

Hello I have 1 question. Is there any application or open source can query data in Delta lake-house ? And return json data to client application, not using delta-sharing. I found https://github.com/rajagurunath/lakehouse-sharing -> it is official solution for lake-house sharing ?

Trying to incrementally upsert changes into a Delta table using autoloader and foreachBatch as described here: https://docs.databricks.com/structured-streaming/delta-lake.html#upsert-from-streaming-queries-using-foreachbatch. It seems my updated records get appended and not updated whatever I try. Should the inital Delta table be set with certain properties? When I use outputMode('Update') in the writeStream I get Data source com.databricks.sql.transaction.tahoe.sources.DeltaDataSource does not support Update output mode.

Hey folks, just a quick question, is the project still considered to be in alpha stage? I am asking that because the classifier for the python bindings is showing as alpha https://github.com/delta-io/delta-rs/blob/e5dd8e2167b94e6856aa531d878584397d5bea69/python/pyproject.toml#L13

Does anyone know of a situation that would cause this error to happen in batch write:

user class threw exception: org.apache.spark.sql.delta.DeltaIllegalStateException: State of the checkpoint doesn't match that of the snapshot.

I found the source code that throws this error, but it doesn't seem possible for it to happen. Clearly it is, and the numberOfFiles check is in there for a reason. Could concurrent inserts cause this? Thanks.

Hello everyone, I have a question related to the ability of applying DDL's on our delta lake catalogs/schemas objects in databricks. I'm looking for a tools like DACPAC (https://learn.microsoft.com/en-us/sql/relational-databases/data-tier-applications/data-tier-applications?view=sql-server-ver16) to automate these changes on the tables from the CI/CD pipelines. any thoughts? thanks

Hello. A stupid question. I have a delta table

which I do not own or created

which is under a strange filepath like:

data///delta_table/

Currently spark does not recognise this, is there any way to work around this without having to change S3 file path?

Will delta lake support vector data(e.g. text embedding or image embedding) ? In general, it needs semantic search indexing like HNSW or ANN.

Hi, I have beginner question regarding delta lake. Am I able to set uuid automatically during merge statement? I've created following code snippet but it does not allow me to use uuid() function.

kv = {c: f"source.{c}" for c in df2.columns}

target_df.alias("target").merge(
    df2.alias("source"), "source.Name = target.Name"
).whenNotMatchedInsert(
    values={
        "uuid": "uuid()",
        **kv
    }
).whenNotMatchedBySourceDelete().execute()

Is there any way to do that or such an action is not supported?

when delta Optimize operation fails in the middle of the process because reasons of low resources for example. does some of the data still get optimized? or is it more transactional in a way that all partitions remains still without any changes?

Hi! I’m quite beginner with Delta but glad to be here! I would ask a little questions, hopefully some of you can give me some light: I’m processing events and storing them in a Delta Table, partitioned by

event_name

and

date

. So the more time I’m ingesting data, the more files and size the table has. Every night I’m running a compaction + vacuum process in order to maintain the table in the best possible way. Compaction (optimize) is perfect because I’m able to filter the partitions I want to optimize but the vacuum command doesn’t allow me to filter the partitions I want to vacuum and does the vacuum of the whole table each time. As I only append data to the table, if we extend this behavior, the vacuum will spend too much time running. For the moment it’s not a problem but I want to anticipate it. Any idea on how can I vacuum/clean the table/old files filtering the partitions I want to vacuum? Or any workaround to avoid too large vacuums? Thanks a lot for your help! Have a nice Thursday! 🙂

Hi all, a quick question, it there a way to add a new column with a default value? in this case is boolean column and i want to set all previous records to false

Hey! Nice to be here. 🙂 I have a question with which I’d love your help: We (a data analytics and science team) are starting to manage our data on Databricks using the medallion architecture. We want to start pulling in data (mainly from MongoDB collections) into DeltaTables, with

delta.enableChangeDataFeed = true

enabled at the silver level, retaining change history. Now, obviously we will have per-column (for the chosen columns) change history from this point onwards. However, for select, very import, columns we do have a way of reconstructing all past value changes from raw data kept at some third data store location (some object storage). We would like to utilize this to get all past changes into said DeltaTables as if the tables were created a few years ago (instead of right now) and were fed all these changes over time, so we can - from this point forwards - get both old and new changes from a single source and using one API. Is this possible? Can we directly inject history into the CDF (Change Data Feed) of a table? The CDF having a schema with a

_commit_timestamp

field points to this possibility (if we can just write rows into the CDF directly). Alternatively, can we write changes to table while overriding the timestamp of the change time instead of it being automatically

now()

? If we can feed the change history from the past up to the present we will end up with the same result. Thank you for reading this, Shay

Hey question about Data Skipping - does collecting statistics happen when writing to delta? or on the optimize job? attaching quotes from databricks delta documentation -

By default Delta Lake on Databricks collects statistics on the first 32 columns defined in your table schema

I have a question about deletion vector metrics. In some of the test files in the master branch, I see metrics for Added and Removed:

"operationMetrics": {
      "numDeletionVectorsAdded": "1",
      "numDeletionVectorsRemoved": "0",

I cannot seem to find any references to

numDeletionVectorsAdded

in the source code, so I am wondering where there is defined?

These metrics are for a DELETE operation, so I expected to find a reference in

DeleteCommandMetrics

I wonder if these metrics are specific to the Databricks implementation, and not in the OSS version?

not sure if this is an usecase for the delta lake, but let me try. We have a set of metadata(json) that we would like to ingest(perhaps through REST) and also consume binary blobs(pdf, zip archives, images, file directories …. perhaps in GB size ) related to the metadata and store it for a given entry. Q1) Is this usecase supported? If yes, Q2) is it possible to version the binary blob data along with the metadata.

facing this issue while installing delta-spark on windows, pyspark is already installed, can anyone guide how to resolve this?

Hello Team, I have a Delta table (version 2.0.1) almost 1.5 PB in size which has never been vacuumed. I was planning to perform cleanup to reduce storage footprint. Would it be advisable to delete the un-referenced files manually (identified by vacuum dry run) since vacuum will block Delta-write operation for hours? Thanks in advance!

How can we install pyspark on windows11 using pip and without hadoop , so that i can use delta spark?

Hey everyone. I'm ambiguous about 2 configuration in delta lake usage. what is the difference between delta.deletedFileRetentionDuration and Retain. for example when we want to apply a retention job on your lake, we have 2 way i guess(I'm not sure they are the same!): First:

delta_table = DeltaTable.forPath(session , path)

delta_table.logRetentionDuration = "interval 10 days"

delta_table.deletes( col("date") < "specified_day" )

session.sql("vacuum delta.path RETAIN 48 HOURS")

session.sql("vacuum delta.path RETAIN 48 HOURS")

Second:

delta_table = DeltaTable.forPath(session , path)

delta_table.logRetentionDuration = "interval 10 days"

delta.deletedFileRetentionDuration = "interval 2 days"

delta_table.deletes( col("date") < "specified_day" )

session.sql("vacuum delta.path")

session.sql("vacuum delta.path")

Are these 2 ways different?!

Hello I have a question regarding the customization of the Delta Parquet Reader in Spark and its effect on byte range queries for Parquet files stored in a distributed environment, specifically on Ceph storage. I'd like to start by asking about the customization of the spark.sql.parquet.columnarReaderBatchSize configuration. In my experiments, I've noticed that modifying this configuration has an impact on memory usage. For instance, setting a very high value results in Out of Memory (OOM) errors. Furthermore, I've observed that, regardless of the value I set for columnarReaderBatchSize (excluding the very high value mentioned earlier), the number of GET queries performed on Ceph remains constant for a given file. Specifically, I consistently observe 18 GET queries. It seems that the columnarReaderBatchSize configuration doesn't affect the number of byte range queries on the distributed storage. Could you shed some light on why this is the case and explain if there are other factors influencing the number of GET queries?

Hi All, I am processing around 130 GB of delta table in EMR on EKS with fargate(Spark Version 3.3.2 and delta 2.2) and it took around 6 hrs for processing. In processing, I selected only a few columns and wrote into another delta table with partitionby on one column with the help of merge function. I tested this job with 16vCPU and 80 GB memory of driver and executor and the number of executors set to 50. So could you please suggest a few optimization techniques which help me to reduce run time. Thanks in advance! #delta-community #delta-oss #deltalake-questions

Could you please suggest a solution? We have a 20GB customer persona in the delta table, and our i*n-house application needs to retrieve customer persona data through an API call*. We require the results to be obtained <*1 second*. Is it possible to create an API on top of the Delta Table for this purpose?

I think I should rather ask this question here. So here it goes.