nvmd - found it - https://wisecubeai.github.io/graphster/#graphster-query

Hi All, I am new here .. my question will sound basic till I ramp up. Looking for support from you all . I am trying to use Spark to write Delta table in S3 and store the meta information in Hive metastore, which can later be queried by Trino . This is how I am starting the pyspark shell -

./pyspark --packages io.delta:delta-core_2.12:2.4.0 \
 --conf "spark.sql.extensions=io.delta.sql.DeltaSparkSessionExtension" \
 --conf "spark.sql.catalog.spark_catalog=org.apache.spark.sql.delta.catalog.DeltaCatalog" \
--conf "fs.s3a.aws.credentials.provide=com.amazonaws.auth.DefaultAWSCredentialsProviderChain" \
--conf "spark.sql.hive.metastore.version=3.1.3" \
--conf "spark.sql.hive.metastore.jars=maven"

This is my spark configuration -

spark = SparkSession.builder \
  .appName("DeltaTableExample") \
  .master("local[*]") \
  .config("spark.hadoop.javax.jdo.option.ConnectionURL", "jdbc:<postgresql://localhost:5435/hive_metastore2>") \
  .config("spark.hadoop.javax.jdo.option.ConnectionDriverName", "org.postgresql.Driver") \
  .config("spark.hadoop.javax.jdo.option.ConnectionUserName", "username") \
  .config("spark.hadoop.javax.jdo.option.ConnectionPassword", "password") \
  .config("spark.sql.warehouse.dir","<s3a://location/hivewarehouse>") \
  .config("spark.sql.extensions","io.delta.sql.DeltaSparkSessionExtension") \
   .config("spark.sql.catalog.spark_catalog","org.apache.spark.sql.delta.catalog.DeltaCatalog") \
   .enableHiveSupport() \
  .getOrCreate()

I am able to write the delta table to S3, but the moment I try to create the table with spark sql to have it’s entry available in Hive metastore

delta_table_path = "<s3a://location/deltatable>"
# Register Delta table in Hive Metastore
spark.sql(f"CREATE TABLE IF NOT EXISTS my_table USING DELTA LOCATION '{delta_table_path}'")

i start getting these errors - I don’t understand from where it is picking this /user/hive/warehouse/my_table , hive warehouse location

23/06/07 00:47:57 WARN HiveExternalCatalog: Couldn't find corresponding Hive SerDe for data source provider delta. Persisting data source table `spark_catalog`.`default`.`my_table` into Hive metastore in Spark SQL specific format, which is NOT compatible with Hive.
00:48:01.034 [Thread-3] ERROR org.apache.hadoop.hive.metastore.RetryingHMSHandler - MetaException(message:file:/user/hive/warehouse/my_table-__PLACEHOLDER__ is not a directory or unable to create one)

Please note I am running hive metastore standalone and hive meta init worked without any erros.

Hii team , i found huge difference between Delta Symlink table and Delta Native table query symlink table took around 17 min of same query but native table took only 19 sec , why symlink table taking more time ?

Hi Team , We are running compaction job at 2am where we first do optimise for current date -1 and then perform vacuum with retain period=2 hours. However the vacuum command doesn’t do anything when we check the S3 folder. I am assuming this is because we have deletedfileretentionduration set as 1 hour and to become the candidate of delete we must perform this vacuum activity after 1 hour of optimise. Can anyone confirm

I receive the following error when creating Delta Lake table with

'delta.minWriterVersion'='6'

table property on Spark 3.4.0 & Delta 2.4.0. Is it a known limitation? Spark 3.3.2 & Delta 2.3.0 had no such failure.

CREATE TABLE a.b (col int) USING DELTA LOCATION '<s3://xxx/xxx>' TBLPROPERTIES ('delta.minWriterVersion'='6')
-> delta.minWriterVersion needs to be one of 1, 2, 3, 4, 5, 7.

Hi team, I have a use case for using the Delta Catalog with Flink 1.17. I can't get the Flink SQL client to create a table akin to the example provided in the connectors repo. Happy to test since this feature was just released. Can I get some help? Btw, it seems like there's some scala version inconsistency.

Hi Team , We are using DeltaLake RunMerge for solving our use case of handling upserts in the data but found out that this utility is taking close to 40-60% time of the whole extraction Load Job , and we are trying to optimise the exectuion time of the job , Can some one help me here ?

Extraction-Load Job Process
1)Read the table data from postgres server in incremental fashion
2)Write data in S3 in parquet format
3)Write the data in AWS-S3  in DeltaFormat
4)Run the Merge to handle upserts

For instance 
If a job is taking 3 min to finish
Step 3 & 4 take close to 1.6-1.7 mins to finish the job

Please find screenshot for more information

https://github.com/delta-io/delta/blob/master/PROTOCOL.md#json-example-2--on-disk-with-absolute-path In which scenario would be advisable to store a deletion vector with the absolute path instead of using the relative path ?

Hello! I am writing some delta lake streaming with pyspark and tables etc. (Databricks and UC otherwise) and want to unit test some streaming functionality locally (SparkSession with warehouse.dir, deltaSparkSessionExtension, spark__catalog, delta__catalog, delta-core...). Is this considered ok to do? I'm just after some input on what to think about/whether this is actually ok to do/dangers/etc. Here's some mock code that runs just fine. Thank you!

does rust package support updates?

Hi everyone, My pyspark job is failing due to no space left on the device. In the job many spill and shuffles are happening due to which the executor fails with space issues. Use case - I am using EMR serverless cluster to process around 130 GB data. In the processing normally it read the delta table from S3 and writes it to another S3 location with partition delta table using the merge function. But it failed due to the lots of shuffles and spill happening on the executor side. I increased the shuffle partitions size but that doesn't help. Even increased the disk size of the executor to 200G along with few spark properties but that also does not help. Delta version -2.2.0 EMR Release version - 6.10.0 Could you please help me with this issue? #delta-community #deltalake-questions

Hi everyone, How can I alter table to add a new column This is the schema I want to add to my delta table

|-- someColumn: map (nullable = true)
 |    |-- key: string
 |    |-- value: struct (valueContainsNull = true)
 |    |    |-- value1: long (nullable = true)
 |    |    |-- value2: string (nullable = true)

I am trying to run the rust example basic_operations and I get this error

Compiling deltalake v0.13.0 (/Users/ma/j/tmp09/delta-rs/rust)
error[E0700]: hidden type for `impl futures::Future<Output = Result<Vec<action::Add>, errors::DeltaTableError>>` captures lifetime that does not appear in bounds
   --> rust/src/operations/write.rs:231:28
    |
231 |   ) -> DeltaResult<Vec<Add>> {
    |  ____________________________^
232 | |     let invariants = snapshot
233 | |         .current_metadata()
234 | |         .and_then(|meta| meta.schema.get_invariants().ok())
...   |
283 | |         .collect::<Vec<_>>())
284 | | }
    | |_^
    |
    = note: hidden type `[async fn body@rust/src/operations/write.rs:231:28: 284:2]` captures lifetime '_#21r

I am trying to run delta-inspect and get this error

Updating <http://crates.io|crates.io> index
error: failed to select a version for `chrono`.
    ... required by package `arrow-array v39.0.0`
    ... which satisfies dependency `arrow-array = "^39.0.0"` of package `arrow v39.0.0`
    ... which satisfies dependency `arrow = "^39"` of package `deltalake v0.13.0 (/Users/ma/j/tmp09/delta-rs/rust)`
    ... which satisfies path dependency `deltalake` of package `delta-inspect v0.8.0 (/Users/ma/j/tmp09/delta-rs/delta-inspect)`
versions that meet the requirements `^0.4.24` are: 0.4.26, 0.4.25, 0.4.24

all possible versions conflict with previously selected packages.

  previously selected package `chrono v0.4.23`
    ... which satisfies dependency `chrono = "^0.4.23"` (locked to 0.4.23) of package `arrow-arith v39.0.0`
    ... which satisfies dependency `arrow-arith = "^39.0.0"` of package `arrow v39.0.0`
    ... which satisfies dependency `arrow = "^39"` of package `deltalake v0.13.0 (/Users/ma/j/tmp09/delta-rs/rust)`
    ... which satisfies path dependency `deltalake` of package `delta-inspect v0.8.0 (/Users/ma/j/tmp09/delta-rs/delta-inspect)`

failed to select a version for `chrono` which could resolve this conflict

Hi team, I get below warning, while ingesting and fetching the data from delta table. i use pyspark 3.3.4 and deltadelta core 2.122.2.0 to ingest data and facing below warnings

23/06/12 04:57:04 WARN DeltaHistoryManager: Found Delta commit 12687 with a timestamp 1686481866687 which is greater than the next commit timestamp 1686306634000.
23/06/12 04:57:04 WARN DeltaHistoryManager: Found Delta commit 12688 with a timestamp 1686481866688 which is greater than the next commit timestamp 1686306757000.
23/06/12 04:57:04 WARN DeltaHistoryManager: Found Delta commit 12689 with a timestamp 1686481866689 which is greater than the next commit timestamp 1686306881000.

1686481866689 - GMT: Sunday, 11 June 2023 111106.689 1686306881000 - GMT: Friday, 9 June 2023 103441 any suggestions how i can resolve these warnings.?

Hi all, I'd like to do data deduplication on a streaming delta table and have the result saved also as a streaming delta table. I've found this guide in Databricks documentation but I don't understand this bit:

In another streaming query, you can continuously read deduplicated data from this Delta table. This is possible because an insert-only merge only appends new data to the Delta table.

The target table will not be a streaming table, as I understand. I can get around this by enabling change data feed and then consuming INSERT operations from the feed. Is this the proposed way? Or is there another way how to get deduplicated streaming table?

Hello Everyone, Im currently trying to use MERGE - WhenMatchedDelete clause like mentioned below. Just wanted to understand what happens if the condition that needs to be specify in the WhenMatchedDelete clause is NONE ? does it delete all the matched records Or It considers None as false in the WhenMatchedDelete condition and drops nothing ? could you please clarify on this ? https://docs.databricks.com/delta/merge.html#language-python

deltaTable = ... # DeltaTable with schema (key, value)

changesDF = spark.table("changes")

latestChangeForEachKey = changesDF \
  .selectExpr("key", "struct(time, newValue, deleted) as otherCols") \
  .groupBy("key") \
  .agg(max("otherCols").alias("latest")) \
  .select("key", "latest.*") \

deltaTable.alias("t").merge(
    latestChangeForEachKey.alias("s"),
    "s.key = t.key") \
  .whenMatchedDelete(condition = "s.deleted = true").execute()

vs

deltaTable.alias("t").merge(
    latestChangeForEachKey.alias("s"),
    "s.key = t.key") \
  .whenMatchedDelete(None).execute()

Hi all, I'm performing a merge statement to implement an scd2 on my delta lake. The source here is a transactional database. I'm detecting that from time to time merge is not working properly. I have some flags called min_flag, max_flag that indicates if the row represents either the first or the last status. However, sometimes when we have updates in our transactional database the merge statement updates correctly the max_flag for the existing row (setting it to 0) but it misses to add the new row that represents the last status. This behaviour is not sistematic, it happens on some days and only for 1 or 2 rows per table, it's very strange. My delta lake version is 2.2.0. Did anyone detect anything similar to this? It doesn't make sense and it's affecting my trust on delta to be honest.

Hi All, We are currently in the process of choosing Delta Lake or Iceberg as the solution to manage our data lake formation. Our data lake consists of both JSON and CSV files, with AWS Glue Catalog managing table definitions for querying CSV data. However, we do not currently manage tables for JSON data. Our objective is to transition to Delta Lake Tables for both JSON and CSV data, with all new data being written to Delta Lake tables. However, we want to avoid the expensive reprocessing of our existing CSV files into Parquet format. Hence, I would like to inquire if there are any known architectural patterns or strategies that can help us query the existing data from the same tables without the need for costly reprocessing of CSV files. We have considered the following possibilities but would greatly appreciate your insights or any additional recommendations: 1. Schema Evolution: Create delta lake tables with CSV data and Evolve schema at some future point to parquet? Is this a possibility? 2. Conversion on Read: Builds on top of #1 but converts CSV data to parquet on read. 3. Conversion: Write pripeline to load csv file in a data frame and than write to parquet tables. We want to avoid this as our lake size is PB in size. Thanks for help. Please feel free to point towards a Readme.

Anyone has experience of persisting data from SQS to Delta in S3? I am considering SQS -> Kinesis -> Spark Streaming or use SQS -> Lambda + Python (delta-rs). Which one do you think is more appropriate? We just need append to the table and I am not sure if delta-rs can handle concurrent write to a table out of the box?

Hi All, recently I've been asked if we can use any database cd tool when deploying our Delta Tables (and also views) to the Databricks workspace. So far wherever I worked with Spark/Hive we didn't defined DDL for our Tables - it was always derived from the set of transformations and finally saved with .saveAsTable(). Now we've got set of Delta Tables, many views that stays on top of them (with relations between the views) and we wanted to automate our deployment. But due to the fact that part of the tables is defined with pyspark transformations and views that depend on these tables are just SQL, I don't know how to connect together these areas and use let's say Flyway. Any ideas? 🙂

Hi all, In this link: https://github.com/delta-io/delta/blob/master/PROTOCOL.md , it contains the log specification. But it keeps being updated. So in the delta lake connector code, I am wondering which version of the log protocol is used? I am asking this because I am trying to implement the delta lake transaction log in Go and the standalone connector is the main source to learn. Thanks in advance!

Hi all, is this a fair comparison between the current state of delta and other table formats? Also, if you know of good articles that do similar comparisons please share, thanks. https://www.onehouse.ai/blog/apache-hudi-vs-delta-lake-vs-apache-iceberg-lakehouse-feature-comparison

I hadn't heard about Hudi until this

hi All , I have spark 3.3.0 trying to use delta lake 2.3.0 (as its compatible with spark 3.3) Because of not having python >=3.7 I created a virtual env of python 3.9 But when trying to install delta-spark== 2.3.0 I get the below error :

/home/bchaudhu/python3_9_env/bin/python3.9  -m pip install delta-spark==2.3.0

Collecting delta-spark==2.3.0
  Using cached delta_spark-2.3.0-py3-none-any.whl (20 kB)
Collecting pyspark<3.4.0,>=3.3.0 (from delta-spark==2.3.0)
  Using cached pyspark-3.3.2.tar.gz (281.4 MB)
  Installing build dependencies ... done
  Getting requirements to build wheel ... done
  Preparing metadata (pyproject.toml) ... error
  error: subprocess-exited-with-error

  Ã Preparing metadata (pyproject.toml) did not run successfully.
  â exit code: 1
  â°â> [80 lines of output]
      /tmp/pip-build-env-d4kbutzi/overlay/lib/python3.9/site-packages/setuptools/dist.py:755: SetuptoolsDeprecationWarning: Invalid dash-separated options
      !!

              ********************************************************************************
              Usage of dash-separated 'description-file' will not be supported in future
              versions. Please use the underscore name 'description_file' instead.

              By 2023-Sep-26, you need to update your project and remove deprecated calls
              or your builds will no longer be supported.

              See <https://setuptools.pypa.io/en/latest/userguide/declarative_config.html> for details.
              ********************************************************************************

      !!
        opt = self.warn_dash_deprecation(opt, section)
      running dist_info
      creating /tmp/pip-modern-metadata-axjhbnal/pyspark.egg-info
      writing /tmp/pip-modern-metadata-axjhbnal/pyspark.egg-info/PKG-INFO
      writing dependency_links to /tmp/pip-modern-metadata-axjhbnal/pyspark.egg-info/dependency_links.txt
      writing requirements to /tmp/pip-modern-metadata-axjhbnal/pyspark.egg-info/requires.txt
      writing top-level names to /tmp/pip-modern-metadata-axjhbnal/pyspark.egg-info/top_level.txt
      writing manifest file '/tmp/pip-modern-metadata-axjhbnal/pyspark.egg-info/SOURCES.txt'
      reading manifest file '/tmp/pip-modern-metadata-axjhbnal/pyspark.egg-info/SOURCES.txt'
      reading manifest template '<http://MANIFEST.in|MANIFEST.in>'
      warning: no previously-included files matching '*.py[cod]' found anywhere in distribution
      warning: no previously-included files matching '__pycache__' found anywhere in distribution
      warning: no previously-included files matching '.DS_Store' found anywhere in distribution
      writing manifest file '/tmp/pip-modern-metadata-axjhbnal/pyspark.egg-info/SOURCES.txt'
      creating '/tmp/pip-modern-metadata-axjhbnal/pyspark-3.3.2.dist-info'
      Traceback (most recent call last):
        File "/home/bchaudhu/python3_9_env/lib/python3.9/site-packages/pip/_vendor/pyproject_hooks/_in_process/_in_process.py", line 353, in <module>
          main()
        File "/home/bchaudhu/python3_9_env/lib/python3.9/site-packages/pip/_vendor/pyproject_hooks/_in_process/_in_process.py", line 335, in main
          json_out['return_val'] = hook(**hook_input['kwargs'])
        File "/home/bchaudhu/python3_9_env/lib/python3.9/site-packages/pip/_vendor/pyproject_hooks/_in_process/_in_process.py", line 149, in prepare_metadata_for_build_wheel
          return hook(metadata_directory, config_settings)
        File "/tmp/pip-build-env-d4kbutzi/overlay/lib/python3.9/site-packages/setuptools/build_meta.py", line 380, in prepare_metadata_for_build_wheel
          self.run_setup()
        File "/tmp/pip-build-env-d4kbutzi/overlay/lib/python3.9/site-packages/setuptools/build_meta.py", line 487, in run_setup
          super(_BuildMetaLegacyBackend,
        File "/tmp/pip-build-env-d4kbutzi/overlay/lib/python3.9/site-packages/setuptools/build_meta.py", line 338, in run_setup
          exec(code, locals())
        File "<string>", line 199, in <module>
        File "/tmp/pip-build-env-d4kbutzi/overlay/lib/python3.9/site-packages/setuptools/__init__.py", line 107, in setup
          return distutils.core.setup(**attrs)
        File "/tmp/pip-build-env-d4kbutzi/overlay/lib/python3.9/site-packages/setuptools/_distutils/core.py", line 185, in setup
          return run_commands(dist)
        File "/tmp/pip-build-env-d4kbutzi/overlay/lib/python3.9/site-packages/setuptools/_distutils/core.py", line 201, in run_commands
          dist.run_commands()
        File "/tmp/pip-build-env-d4kbutzi/overlay/lib/python3.9/site-packages/setuptools/_distutils/dist.py", line 969, in run_commands
          self.run_command(cmd)
        File "/tmp/pip-build-env-d4kbutzi/overlay/lib/python3.9/site-packages/setuptools/dist.py", line 1244, in run_command
          super().run_command(command)
        File "/tmp/pip-build-env-d4kbutzi/overlay/lib/python3.9/site-packages/setuptools/_distutils/dist.py", line 988, in run_command
          cmd_obj.run()
        File "/tmp/pip-build-env-d4kbutzi/overlay/lib/python3.9/site-packages/setuptools/command/dist_info.py", line 104, in run
          bdist_wheel = self.get_finalized_command('bdist_wheel')
        File "/tmp/pip-build-env-d4kbutzi/overlay/lib/python3.9/site-packages/setuptools/_distutils/cmd.py", line 304, in get_finalized_command
          cmd_obj = self.distribution.get_command_obj(command, create)
        File "/tmp/pip-build-env-d4kbutzi/overlay/lib/python3.9/site-packages/setuptools/_distutils/dist.py", line 860, in get_command_obj
          klass = self.get_command_class(command)
        File "/tmp/pip-build-env-d4kbutzi/overlay/lib/python3.9/site-packages/setuptools/dist.py", line 989, in get_command_class
          self.cmdclass[command] = cmdclass = ep.load()
        File "/tmp/pip-build-env-d4kbutzi/overlay/lib/python3.9/site-packages/setuptools/_vendor/importlib_metadata/__init__.py", line 208, in load
          module = import_module(match.group('module'))
        File "/usr/local/lib/python3.9/importlib/__init__.py", line 127, in import_module
          return _bootstrap._gcd_import(name[level:], package, level)
        File "<frozen importlib._bootstrap>", line 1030, in _gcd_import
        File "<frozen importlib._bootstrap>", line 1007, in _find_and_load
        File "<frozen importlib._bootstrap>", line 986, in _find_and_load_unlocked
        File "<frozen importlib._bootstrap>", line 680, in _load_unlocked
        File "<frozen importlib._bootstrap_external>", line 850, in exec_module
        File "<frozen importlib._bootstrap>", line 228, in _call_with_frames_removed
        File "/tmp/pip-build-env-d4kbutzi/overlay/lib/python3.9/site-packages/wheel/bdist_wheel.py", line 28, in <module>
          from .macosx_libfile import calculate_macosx_platform_tag
        File "/tmp/pip-build-env-d4kbutzi/overlay/lib/python3.9/site-packages/wheel/macosx_libfile.py", line 43, in <module>
          import ctypes
        File "/usr/local/lib/python3.9/ctypes/__init__.py", line 8, in <module>
          from _ctypes import Union, Structure, Array
      ModuleNotFoundError: No module named '_ctypes'
      [end of output]

  note: This error originates from a subprocess, and is likely not a problem with pip.
error: metadata-generation-failed

à Encountered error while generating package metadata.
â°â> See above for output.

note: This is an issue with the package mentioned above, not pip.
hint: See above for details.

Hi! Does https://github.com/delta-io/delta-rs (Python version) guarantee the delta-concurrency ?

Hi everyone!! At my company, we are trying to use the delta lake with AWS and with Databricks. I am not a fan of Databricks so I do everything with AWS Glue but the compatibility with Delta seems strange. I have a hard time using the spark-delta with Glue 3 (so no interactive notebooks for me) because I don’t seem to be able to set it up. It works fine with Glue 4. But I was wondering how do other people do it? How do you work with delta and Glue? If you use a crawler, what are the advantages of using native tables or creating a symlink? How do you interact with your delta tables? I can use Athena of course but it is read only and does not support time travel. For now this has been fine but it kind of makes one of the cool things about delta a bit unusable (time travel). Maybe I am just doing it all wrong, can anyone comment on their experiences and tricks to work with #deltalake-on-aws ? Thank you 😊

Hello All, I am a Graduate Student Currently Researching on Delta Lake, I have few questions regarding the lakehouse architecture. 1) Can we consider Lakehouse as third generation Cloud Data Warehouse ? if so why didnt the authors mention, that point explicitly over the delta lake .. https://www.vldb.org/pvldb/vol13/p3411-armbrust.pdf Nor in the following one https://cs.stanford.edu/~matei/papers/2021/cidr_lakehouse.pdf Therefore, can anyone from team confirm, is this inference valid or is it, any different from my perception. Sorry for positng this question over here, please neglect if this academic question is irrelevant.

Hello Team, I am here with another question, Soo Deltalake is created to perform, incremental updates over cloud object stores that ensures ACID properties. As per the literature it mentions that, Delta lake took reference from hudi. In 2021 lakehouse architecture got published, but my question here is 1) Lakehouse combines the best worlds of both Datalake and Warehouse then whats the need of integrating with Snowflake (A Traditional Datawarehouse) When Photon Query engine, can do what the Snowflake does. Therefore plugging BI tools to visualize data directly from lakehouse is a viable option ? (As per the literature) This raises my above question again ! Is this for data governance , data auditing, data lineage purposes ? if so we can plugin data governance tools like Openmetadata to visualize and track everything directly from lakehouse ! Or else is this related to lack of support offered to lakehouse as of now ? Let me know your thoughts on this 🙂

I have an issue with delta that is only reproducible inside a docker container. When I run

spark-submit

on my local host I have no errors. However, I wanted to test this spark application in docker, which I guess is still technically my local host, but I want to submit the program from the docker container. I have four containers. spark-master, two spark-workers, spark-history-server, and a spark-driver. All containers are on the same docker network. Inside the spark-driver container is where I launch

spark-submit --properties-file ./src/spark/spark-defaults.conf ./src/start_pipeline.py

I receive the following error

An error occurred while calling o515.load.

: java.util.concurrent.ExecutionException: org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 11.0 failed 4 times, most recent failure: Lost task 0.3 in stage 11.0 (TID 156) (172.23.0.3 executor 3): org.apache.spark.SparkFileNotFoundException: File file:/opt/ufo-lakehouse/lakehouse/ufo/bronze/_delta_log/00000000000000000000.json does not exist

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.

This file does exist, the spark program created it! I first thought it may be permission issues so I set all my folders/files with permission 777 but still have the same error. Any help or guidance would be much appreciated. Thank you!