Hi all, has anybody tried to run delta lake unit tests in Docker? I am using pytest, and was following this guide https://docs.delta.io/latest/quick-start.html#set-up-project but the delta package won't load. On

DeltaTable.isDeltaTable

I get this error:

object is not callable

. Thanks!

Hi all, I am using concat_Ws in Databricks notebook. Is the concat_Ws function deterministic? Does it return same order in every run?

Hey guys, im a little confused and maybe you can help me with that. Im currently experimenting with delta lake on minio and i love the simplicity of delta tables with direct path´s for my SQL commands. Now i´ve tried to switch my examples to delta table names and namespaces. Everything is fine so long as i create my table within a spark session and play around with it. If i start another session and try to read the table by name, i get always an

AnalysisException: Table or view not found: default.delta_table;

doesnt matter if i use "default" as namespace or my own name... (Yes the table data and deltalog is there in minio, so everything is fine from storage perspective and i can read the table by path) Could you explain why i can only see my tables during a session and maybe what i could do to see my tables also from other sessions as well?

Hello everyone, From a list of folders on s3, how can I identify from Databricks which ones are Delta tables which are not?

I was performing benchmarking for open table formats against some parameters. I created two tables, one is in iceberg format(parquet) and another in delta lake using spark-3.3(emr-6.9). In the initial data load (insert) into the table operation, iceberg output data volume was 102GBs against input data volume of 161GBs. While on the delta lake the output data volume was 228GBs against the same input data volume of 161GBs. Do anyone know the reason?

Hey, I'm trying to use Delta Live, we have a python wheel library. all notebooks should use this lib. How to install that lib in the Delta Live clusters. Thanks

Is there a way we can define isolation level while performing writes on delta tables? I read somewhere that three isolation levels are supported i.e. Serializable, WriteSerializable and Snapshot.

Hi team - we have managed delta tables in unit catalog, how can I share them into Snowflake? https://docs.delta.io/latest/snowflake-integration.html#language-sql I was trying to follow instructions over here but as tables are managed not external gives this error -

Error in SQL statement: UnityCatalogServiceException: [RequestId= ErrorClass=INVALID_PARAMETER_VALUE.LOCATION_OVERLAP] Input path url 's3://<path>' overlaps with managed storage

while running this command-

GENERATE symlink_format_manifest FOR TABLE delta.`<path-to-delta-table>`

is there any way available to generate manifest and share those into snowflake if delta tables under unity catalog?

Hi all, I have a delta table. When I drop the table, the files in the mount (Azure Data Lake Storage) was not cleaned. Is there any drop table option which also cleans the file contents? A command like DROP TABLE DELTA_TABLE1 CASCADE

Hi Team - I have a write focused delta table with rust driver. I would like to set the compression to "zstd" but can't seem to find how to write a batch with that set up? Is it supported? Thanks!

Hi, anyone has tried to read delta lake tables using athena? Getting an HIVE_CANNOT_OPEN_SPLIT error even after optimizing the table

Hi deltausers, Situation: We have the need to load in empty tables without providing a schema at first, before another pipeline actually kicks off to write data to this location (using the delta format of course). For this to work we need to set the spark.conf.set(“Spark.Databricks.delta.schema.autoMerge”,”true”) so that schema evolution is enabled when the pipeline writes data to this empty location that has been created from step one. However, upon writing data to the empty location in step two, I am still getting an error saying that I need to set the save mode as overwrite and also need to specify option(overwriteSchema) to true. Is there no way that these two above options can be set at the cluster level so that we don’t affect our current delta writer which is being used by hundreds of other pipelines? Is my approach correct?

Hi team, I am using delta 2.1.1 to perform optimization with z-order. This is my function:

def zorder(spark: SparkSession, opts: AppOptions): Unit = {
    val path = opts.getString("path")
    val partitionCondition = opts.getString("partition_condition")
    val optimizeCols = opts.getString("optimize_cols").split(",").map(_.trim)

    val table = DeltaTable.forPath(spark, path)
    table.optimize()
      .where(partitionCondition)
      .executeZOrderBy(optimizeCols: _*)

    spark.conf.set("spark.databricks.delta.retentionDurationCheck.enabled", false)
    spark.sql(s"ALTER TABLE delta.`${path}` SET TBLPROPERTIES('delta.deletedFileRetentionDuration' = 'interval 1 minute')")
    spark.sql(s"ALTER TABLE delta.`${path}` SET TBLPROPERTIES('delta.logRetentionDuration' = 'interval 1 day')")
    Thread.sleep(90000) // sleep 1.5 minutes before vacuum
    table.vacuum()
  }

The problem is that the table I need to optimize has a lot of small files, around 400 files due to a batch job that writes every 15 minutes and then partitions by 3 (having 3 sources). When I try to optimize with z-order, even uses a lot of resources and I still get the following issue: "Caused by: org.apache.spark.SparkException: Job aborted due to stage failure: ResultStage 15 (run at ForkJoinTask.java:1402) has failed the maximum allowable number of times: 4. Most recent failure reason: org.apache.spark.shuffle.MetadataFetchFailedException: Missing an output location for shuffle 2 partition 0".

Hi team, I'm in spark streaming application context and I notice that job seems to list all transactions files contained in '_delta_log' before beginning to compute snapshot. Is it the expected behavior or I missed something ? In my mind, to build a snapshot, it only need to get most recent checkpoint file thanks to 'last_checkpoint' file and read from it and go on with the few others *.json above, am I right ?

Hello everyone! Just a doubt regarding Delta-Rs python library. There are any plans to include on this API a way to handling the transaction log directly? For an instance add/remove files like we do in Rust? Thanks!

Hi, Any idea why the overwrite mode when set with option does not work with delta format. However when it is set using

mode(“overwrite”)

it works fine.

Hi Team, I am looking for an answer around can we write to a delta table when we are doing vacuum on it ? I got conflicting results in search and havent found a doc which says that we should write simultaneously or not. can anyone help me with a link where standard practice is recommended. we are trying to build a daily vacuum and optimize job and debate is can we write or do we have to pause streams to write to that table

Do we know when will Delta support Identity columns ? I am using Synapse(Delta 2.2) and see that currently Identity columns are not supported. Thinking to use monotonically increasing function>However, the approach has its own challenges, so thinking if the feature can be made available, it will make life easy :)

Hi Team, according to understand some behaviors about delta table reading, I tried to just read a delta table and monitor it using the 'mc watch' command on my minio, I notice that some requests seems to be duplicated.

I use the docker image 'all-spark-notebook' that launch spark in local mode with PySpark==3.3.2 and delta-spark==2.2.0 and a minio docker image, all wrapped with docker-compose

Is it expected behavior ?

Hi All, I am new to delta lake and I am trying to upsert data to a delta table.The scenario is as below: target.alias("target").merge( source.alias("source") , "target.partition_column in (partition_value) and target.id = source.id) .whenMatchedUpdateAll() .whenNotMatchedInsertAll() .execute()) A very simple partition pruning followed by upsert. The target is partitioned into two partitions. Now my merge operation is creating only one file in the partition after write. And in the spark ui all the data is shuffled into one executor which is then doing all the writing, which is not very efficient. Can someone please help me on this I am using OSS delta lake.

In Databricks, i am using pyspark to read CSV. I need to skip the first 2 rows. My csv header is at 3rd row. I could not find a way to do it in pyspark (i can only see this skiprows option in Pandas). Really appreciate if any body could help. Thanks.

Is there a way to purposefully aquire a lock on a delta table?

can we copy delta table using distcp to a different cluster ? or there is any other better way of copying the complete delta table ?

Hi All, we recently published a blog on CDF Failure scenarios and contributed the code to the jodie library https://medium.com/@joydeep.roy/change-data-feed-time-travel-failure-scenarios-prevention-recovery-5606c65d0c2e Please check it out and provide feedback/comment

Hi Team follow up question on my vacuum question earlier. If my delta table lets say x has table property set for retention to be 1 day when I run a vacuum saying vacuum table x retain 30 days what happens in this scenario ? am I missing an opportunity on cleaning up some files that are not being used ? or the table itself will not have it as it has only 1 day retention set

Is it possible to retain certain points in time using Vacuum command? For example always retain the data for the first of the month.

Hello

How we can archive to different level of S3 tier ?

Hi everyone, I am used to work woth architectures where I use adf to ingest data(database, sftp, s3, etc.) and databricks to process data. Is there an existing architecture where databricks is used for ingestion and process and can somebody share a link or picture?

Hi folks. I ran zorder and compact on a file of 1MB, but found that the job processed 10k partitions in one of the stages. Is there any way to configure the number of partitions at this point?