delta-lake #deltalake-questions

sabari dass

02/02/2023, 8:05 AM

Also want to know what will happen if we don’t have checkpoints?

sabari dass

02/02/2023, 8:06 AM

Can anyone please help me on this?

Roberto

02/02/2023, 2:58 PM

Hi All, I have a question related to merge command. Is possible when matched to use the .updateExpr to sum the value for the original table with the new value?

Vishal Kadam

02/03/2023, 6:58 PM

Hi All, Is the delta merge function async? Means it will execute the next line of code without waiting for the merge operation to complete.

Vivek B

02/04/2023, 5:44 AM

Hi All, can you provide a way to fetch OSS delta lake data through low latency Rest APIs

Lennart Skogmo

02/05/2023, 9:33 AM

Does anyone have any tips about the best way to atomically replace the content of a delta table? With other tech you have some options like, transaction truncate & insert, insert into new table and swap tables with atomic rename, and ofc merge into if we have keys.

Rahul Sharma

02/05/2023, 1:39 PM

Hii Team , i am getting file not found exception ,The default retention threshold for the Delta files is set to 72 hours.

Sanjeeb Dey

02/06/2023, 5:41 AM

Hi All, I m using Spark 3.3.0 and Delta 2.2.0

Sanjeeb Dey

02/06/2023, 5:43 AM

facing an issue (not sure if it one), but whenever I do a merge logic, the number of partitions in spark processing reduces to equivalent to the table partition

Sanjeeb Dey

02/06/2023, 5:43 AM

If anyone has faced the same, can we have more partitions?

Alex

02/06/2023, 9:08 AM

Copy code

It is possible the underlying files have been updated. You can explicitly invalidate the cache in Spark by running 'REFRESH TABLE tableName' command in SQL or by recreating the Dataset/DataFrame involved.

How it can be possible to have a file name cached in spark? We are getting a strange error regarding:

Copy code

java.io.FileNotFoundException: No such file or directory: s3a://.....snappy.parquet

I restarted the app but still the same issue. I checked delta logs and the file is not present in delta_log. How is spark able to remember this file when it is not even present in delta log?

Alex

02/06/2023, 9:34 AM

Seems like some files are stored in .checkpoint.parquet and spark is trying to read those files from delta. But i don't understand how is it possible to have those files in the checkpoint parquet but those files doesn't exist in the table. How is it possible?

Morgan

02/06/2023, 10:59 AM

Hi All, I'm proceeding some tests about log retention and I intentionally put the value of 'logRetentionDuration' at "interval 0 days" and I'm proceeding with append only on the current delta table, but the transaction log files never removed ? Did I missed something ? (Delta 2.1.1)

sabari dass

02/06/2023, 11:07 AM

Hi All, Is there anyway to get next day timestamp value from current timestamp? For date field, I can do it by using sf.date_add() function but not sure about timestamp field. Can anyone please help me on this? Thanks!

Álvaro José Baranoski

02/06/2023, 2:33 PM

Hello everyone, I am running some tests on my local machine with a Spark mini cluster (4 docker containers, 3 workers and 1 master). I'm trying to create a simple delta table sending commands to the cluster with

spark-submit

. The table that I'm trying to create is the very first one in the getting started section, like so:

Copy code

data = spark.range(0, 5)
data.write.format("delta").save("/tmp/delta-table")

Whoever, when executing the command, the log presents the following error:

Copy code

[2023-02-06, 11:09:36 -03] {spark_submit.py:495} INFO - 23/02/06 11:09:36 WARN TaskSetManager: Lost task 0.0 in stage 1.0 (TID 2) (172.19.0.3 executor 2): java.io.FileNotFoundException:
[2023-02-06, 11:09:36 -03] {spark_submit.py:495} INFO - File file:/home/alvaro/airflow/tmp/delta-table/_delta_log/00000000000000000000.json does not exist
[2023-02-06, 11:09:36 -03] {spark_submit.py:495} INFO - 
[2023-02-06, 11:09:36 -03] {spark_submit.py:495} INFO - It is possible the underlying files have been updated. You can explicitly invalidate
[2023-02-06, 11:09:36 -03] {spark_submit.py:495} INFO - the cache in Spark by running 'REFRESH TABLE tableName' command in SQL or by
[2023-02-06, 11:09:36 -03] {spark_submit.py:495} INFO - recreating the Dataset/DataFrame involved.

What should I do to get the delta table to be created on my local machine? Is it possible to do so using this kind of Spark cluster? If not, what is the best way to Spark + Delta Lake + Airflow on my local machine? Thanks in advance!!

Edmondo Porcu

02/06/2023, 6:22 PM

We are facing a dilemma around how to improve the layout on disk of data which is going to be queried by using lte/gte filters on a decimal value. This sounds like a great use case for Z-Ordering optimizations in Delta Lake, but we are also concerned that this might result in an extraordinary amount of IO operations or shuffling, and we considered to create our own “bin” column and partition by bin. We haven’t found much on Delta Lake documentation that compares Z-Orders with explicit binning, does anyone have some useful insight on the topic?

Kenny Ma

02/06/2023, 10:34 PM

@Krzysztof Chmielewski Do you have any performance and load benchmark that you can share with us for the Flink Delta connector? Also do you have docs that I can read more about Delta connector during job submission? I saw in the Flink client log that it is trying to read in a bunch of rows and I am trying to understand why it is doing that in the client.

Copy code

2023-02-06 14:28:14,286 INFO  io.delta.standalone.internal.DeltaLogImpl                    [] - Loading version 34082 starting from checkpoint 34080.
2023-02-06 14:28:14,305 INFO  io.delta.standalone.internal.SnapshotImpl                    [] - [tableId=6ee75af7-8e09-4d40-b92a-2cb10a844f20] Created snapshot io.delta.standalone.internal.SnapshotImpl@4202bfe8
2023-02-06 14:28:15,756 INFO  org.apache.hadoop.fs.s3a.S3AInputStream                      [] - Switching to Random IO seek policy
2023-02-06 14:28:16,117 INFO  shadedelta.org.apache.parquet.hadoop.InternalParquetRecordReader [] - RecordReader initialized will read a total of 3094160 records.
2023-02-06 14:28:16,117 INFO  shadedelta.org.apache.parquet.hadoop.InternalParquetRecordReader [] - at row 0. reading next block
2023-02-06 14:28:24,143 INFO  org.apache.hadoop.io.compress.CodecPool                      [] - Got brand-new decompressor [.snappy]
2023-02-06 14:28:24,210 INFO  shadedelta.org.apache.parquet.hadoop.InternalParquetRecordReader [] - block read in memory in 8093 ms. row count = 639939
2023-02-06 14:28:28,277 INFO  shadedelta.org.apache.parquet.hadoop.InternalParquetRecordReader [] - Assembled and processed 639939 records from 38 columns in 3861 ms: 165.74437 rec/ms, 6298.286 cell/ms
2023-02-06 14:28:28,278 INFO  shadedelta.org.apache.parquet.hadoop.InternalParquetRecordReader [] - time spent so far 67% reading (8093 ms) and 32% processing (3861 ms)
2023-02-06 14:28:28,278 INFO  shadedelta.org.apache.parquet.hadoop.InternalParquetRecordReader [] - at row 639939. reading next block
2023-02-06 14:28:36,215 INFO  shadedelta.org.apache.parquet.hadoop.InternalParquetRecordReader [] - block read in memory in 7935 ms. row count = 629616
2023-02-06 14:28:39,398 INFO  shadedelta.org.apache.parquet.hadoop.InternalParquetRecordReader [] - Assembled and processed 1269555 records from 38 columns in 7018 ms: 180.89983 rec/ms, 6874.1934 cell/ms
2023-02-06 14:28:39,399 INFO  shadedelta.org.apache.parquet.hadoop.InternalParquetRecordReader [] - time spent so far 69% reading (16028 ms) and 30% processing (7018 ms)
2023-02-06 14:28:39,400 INFO  shadedelta.org.apache.parquet.hadoop.InternalParquetRecordReader [] - at row 1269555. reading next block
2023-02-06 14:28:47,303 INFO  shadedelta.org.apache.parquet.hadoop.InternalParquetRecordReader [] - block read in memory in 7902 ms. row count = 638410

👀 1

Aaron Kub

02/07/2023, 2:27 AM

Hello fellow data people. :) I'm encountering what seems to be a bug with the vacuum command. We have a large delta table and want to preserve 21 days of version history. I've set the

delta.deletedFileRetentionDuration

table property to "interval 21 days". When we run the vacuum command with a 504 hour retention period (21 days) we end up with corrupted snapshots that were created within the 21 day period. By "corrupted" I mean that when a version that is, say, 10 days old is read there is a

FileNotFoundException

for certain part files that have now been deleted unexpectedly. After digging deeper by running vacuum commands in dry-run mode I've concluded that the deleted files were (1) "removed" in a snapshot version greater than 7 days ago and (2) created over 21 days ago. The ultimate culprit seems to be that the snapshot that Delta uses to generate the validFiles object in the

VaccumCommand.gc

method is filtering out files that were removed more than 7 days ago because it's defaulting to the default retention period. I've confirmed that the snapshot only has 7 days of removed files by doing something like this:

Copy code

val deltaLog = DeltaLog.forTable(spark, "<s3://path-to-table|s3://path-to-table>")
val snapshot = deltaLog.update()
val removedFiles = snapshot.state
	.filter(_.remove != null)
	.collect()
	.map(_.unwrap.asInstanceOf[RemoveFile])
	.sortBy(_.delTimestamp)

Lastly, to double check that the default

TOMBSTONE_RETENTION

value is in fact what is causing this behavior, I cloned the delta project and changed the default value to 21 days, then ran the above steps again. I then had 21 days of removed file history in the snapshot. Sanity checks: I am completely sure that the retention period on the table is indeed set to 21 days by using the

SHOW TBLPROPERTIES

command. The checkpoint files in the

_delta_log

directory also contain 21 days of removed file history. It's just the in-memory snapshot that filters the removed files down to the most recent 7 days. We are using AWS EMR, Scala 2.12, and Delta 1.0.0, although I have reproduced this behavior on delta 2.1.0. It feels like I must be missing something. Has anyone encountered this?

Grainne B

02/07/2023, 10:45 PM

Hi all! I am trying to run a simple spark script in Docker to create a delta table in aws, wondering what Docker image people are using and would recommend ? I have tried a few different ones but I run into version issues with hadoop-aws and delta-core !! I am using Mac m1 , so the image would have to be compatible (arm64) Edit - I have been using image apache/spark-py:v3.3.0 , but I need to install hadoop separately .. which is a pain

Dhruvil Shah

02/08/2023, 1:15 AM

Hi I am trying to enable Delta Table CDC but in AWS Glue but didn't work as expect it.

Dhruvil Shah

02/08/2023, 1:19 AM

Copy code

import sys
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.job import Job
 
## @params: [JOB_NAME]
args = getResolvedOptions(sys.argv, ['JOB_NAME'])
 
sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args['JOB_NAME'], args)
 
df = spark.createDataFrame([(1, "foo"),    (2, "bar"), ],["id", "label"])
df.write.format("delta").mode("overwrite").save("<s3://lakeformation-dhruv/cdctest/>")
print( "Count ")
job.commit()

Dhruvil Shah

02/08/2023, 1:19 AM

Copy code

import sys
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.job import Job

## @params: [JOB_NAME]
args = getResolvedOptions(sys.argv, ['JOB_NAME'])
a
sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args['JOB_NAME'], args)


df = spark.read.format("delta") \
  .option("readChangeFeed", "true") \
  .load("<s3://lakeformation-dhruv/cdctest/>")


print ( df )
print( df.show() )


job.commit()

Dhruvil Shah

02/08/2023, 1:21 AM

Can someone help with it?

Rahul Sharma

02/08/2023, 5:12 AM

Hello Team ,i have schedule a Deltalake job raw and refine 10 days ago and i have set delta file and log retention duration is 5 days ,i am getting below error please can someone suggest me to change the retention duration default or this error is related to something else

Copy code

"pyspark.sql.utils.StreamingQueryException: The stream from your Delta table was expecting process data from version 142,\nbut the earliest available version in the _delta_log directory is 159. The files\nin the transaction log may have been deleted due to log cleanup. In order to avoid losing\ndata, we recommend that you restart your stream with a new checkpoint location and to\nincrease your delta.logRetentionDuration setting, if you have explicitly set it below 30\ndays.\nIf you would like to ignore the missed data and continue your stream from where it left\noff, you can set the .option(\"failOnDataLoss\", \"false\") as part\nof your readStream statement.\n=== Streaming Query ===\nIdentifier: JWR-SEC-

Aaron Kub

02/08/2023, 5:46 PM

Does anyone know when version 2.3.0 will be released?

Sekhar Sahu

02/08/2023, 9:58 PM

Hi Folks.. I am trying out delta lake on AWS EMR 6.9.0 release label following the documentation listed here. When I try to run the following, I get this error.. Can anyone please help debug this?

Copy code

23/02/08 21:53:36 WARN HiveExternalCatalog: Couldn't find corresponding Hive SerDe for data source provider delta. Persisting data source table `default`.`delta_table` into Hive metastore in Spark SQL specific format, which is NOT compatible with Hive.
Traceback (most recent call last):
  File "<stdin>", line 4, in <module>
  File "/usr/lib/spark/python/pyspark/sql/session.py", line 1034, in sql
    return DataFrame(self._jsparkSession.sql(sqlQuery), self)
  File "/usr/lib/spark/python/lib/py4j-0.10.9.5-src.zip/py4j/java_gateway.py", line 1322, in __call__
  File "/usr/lib/spark/python/pyspark/sql/utils.py", line 196, in deco
    raise converted from None
pyspark.sql.utils.IllegalArgumentException: Can not create a Path from an empty string

Code:

Copy code

## Create a DataFrame
data =  spark.createDataFrame([("100", "2015-01-01", "2015-01-01T13:51:39.340396Z"),
("101",  "2015-01-01", "2015-01-01T12:14:58.597216Z"),
("102", "2015-01-01", "2015-01-01T13:51:40.417052Z"),
("103",  "2015-01-01",  "2015-01-01T13:51:40.519832Z")],
["id", "creation_date",  "last_update_time"])

## Write a DataFrame as a Delta Lake dataset to the S3  location
spark.sql("""CREATE  TABLE IF NOT EXISTS delta_table (id string, creation_date string, 
last_update_time string)
USING delta location
'<s3://DOC-EXAMPLE-BUCKET/example-prefix/db/delta_table>'""");

data.writeTo("delta_table").append()

pyspark command used

Copy code

pyspark --master yarn --deploy-mode client --repositories <http://repo.hortonworks.com/content/groups/public/,https://repos.spark-packages.org/,https://oss.sonatype.org/content/repositories/snapshots>  --conf spark.sql.adaptive.coalescePartitions.initialPartitionNum=5000 --conf spark.databricks.delta.optimize.maxFileSize=250000 --conf spark.sql.extensions=io.delta.sql.DeltaSparkSessionExtension --conf spark.executor.extraJavaOptions=-XX:+UseG1GC --conf spark.driver.maxResultSize=0      --conf spark.sql.catalog.spark_catalog=org.apache.spark.sql.delta.catalog.DeltaCatalog --conf spark.databricks.delta.optimize.repartition.enabled=true   --conf spark.databricks.delta.autoOptimize=true  --packages io.delta:delta-core_2.12:2.1.0

Leandro Rouberte

02/09/2023, 12:35 AM

Hi, team. I'm working on setting multi-cluster writing for my delta lake using dynamodb. I have a scenario where a spark job running in an aws account will need to access the dynamodb table living in another account. I see that the options on the doc only accepts table name and region. Is there a way to pass the account id?

gopinath s

02/09/2023, 11:39 AM

Hi Team, when i start the delta sharing server using the below command, its not starting properly. Please help. 1877 [armeria-boss-http-*:8080] INFO com.linecorp.armeria.server.Server - Serving HTTP at /[0000000:0%0]:8080 - http://127.0.0.1:8080/ root@Ubuntu:/delta/delta-sharing-server-0.5.4# curl http://localhost:8080/ 404 Not Found

Roberto

02/09/2023, 12:06 PM

Good morning team, I have a question regarding the vacuum command. Could execute at the same time vacuum and merge in the same table (two separate processes)?

Nate Kuhl

02/09/2023, 5:13 PM

I have a Spark Structured Streaming application that is merging data into a target Delta Lake table using UpdateAll every microbatch. I had

mergeSchema=True

set on the DataFrameWriter and was surprised to see that the application wasn’t progressively merging in new columns. I looked up some documentation and noticed that for merges I need to set

spark.databricks.delta.schema.autoMerge.enabled=true

to enable schema evolution. I want to make sure I’m understanding the two configurations correctly: • mergeSchema=true will enable schema evolution for write operations •

spark.databricks.delta.schema.autoMerge.enabled=true

enables schema evolution for merge operations (InsertAll and UpdateAll)