delta-lake #deltalake-questions

Anmol Rastogi

05/05/2023, 7:55 AM

hi #deltalake-questions I am trying to read a deltalake table using pure python and trying to create a pipeline using pandas :

Copy code

"""
full_data_load.py
~~~~~~~~~~~
This Python module is used to load data from RDS to data lake storage (i.e. S3 with respect to AWS)
"""

import os
import deltalake


env = 'dev'
region = 'ap-south-1'

os.environ["ENV"] = env
os.environ["REGION"] = region

bucket = 'analytics-pipeline-ap-south-1'


def main():
    data_read = deltalake.table.DeltaTable(f's3://{bucket}/DeltaLake_convert/Anmol', storage_options={'region': region})
    print(data_read)


# entry point for ELT application
if __name__ == '__main__':
    main()

while doing this i am getting following error: deltalake.PyDeltaTableError: Not a Delta table: No snapshot or version 0 found, perhaps s3://{bucket}/DeltaLake_convert/Anmol is an empty dir? while if you see below image the deltalog directory and parquet exists there. am i doing something wrong? Note: I am not looking for something with spark. 🙂

Rahul Sharma

05/05/2023, 8:06 AM

hello Team , i am getting below error while reading delta table data using athena external table from manifast loc

Joydeep Banik Roy

05/05/2023, 9:59 AM

Hi delta io guys Please check out our new blog on Delta Merge Optimization:- https://www.linkedin.com/posts/joydeep-banik-roy_delta-mergeoptimisation-strategies-a[…]867588489216-WDby?utm_source=share&utm_medium=member_desktop

👀 1

Joydeep Banik Roy

05/05/2023, 10:00 AM

Let me know your thoughts

sharath

05/05/2023, 11:10 AM

Hello.. When we update data of one column.. does delta in the backend update only that column or the whole row is re-written ?

Abidi Gassen

05/05/2023, 6:05 PM

I hope you enjoy the read 😃 😃 Any feedback would be much appreciated https://www.linkedin.com/posts/databeans_mergeschemaevolution-activity-7060311422156566529-8MyV?utm_source=share&utm_medium=member_ios

kamal kaur

05/05/2023, 7:48 PM

Hi all - Delta partitioning question - We are using open-source Delta on S3. Is it okay to partition delta table by an

id

with around 500k unique values. We are using merge operation in that table using

id

and we will be querying using Trino by that

id

. I know those are too many partitions but if the use case is just to query/merge that table by

id

, will this work?

Rambabu Posa

05/05/2023, 8:12 PM

Hi, I’m trying to covert a

parquet

file to

delta

using the below code snippet

Copy code

public static void main(String[] args){

        SparkConf conf = new SparkConf();
        conf.setAppName("Parquet to Delta Converter");
        conf.setMaster("local[*]");

        SparkSession session = SparkSession.builder()
                .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")
                .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog")
                .config(conf)
                .getOrCreate();

        DeltaTable.convertToDelta(session, "parquet.`table/betting`");

        session.stop();
    }

All my parquet files are available at

table/betting

. but getting this error:

Copy code

java.lang.NoSuchMethodError: org.apache.spark.sql.execution.datasources.parquet.ParquetToSparkSchemaConverter$.$lessinit$greater$default$4()Z
	at org.apache.spark.sql.delta.commands.ParquetTable.$anonfun$mergeSchemasInParallel$2(ConvertToDeltaCommand.scala:595)
	at org.apache.spark.rdd.RDD.$anonfun$mapPartitions$2(RDD.scala:855)
	at org.apache.spark.rdd.RDD.$anonfun$mapPartitions$2$adapted(RDD.scala:855)
	at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
	at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:365)
	at org.apache.spark.rdd.RDD.iterator(RDD.scala:329)
	at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90)
	at org.apache.spark.scheduler.Task.run(Task.scala:136)
	at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:548)
	at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1504)
	at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:551)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
	at java.lang.Thread.run(Thread.java:748)

Many thanks

Ujjwal Kumar Gupta

05/07/2023, 2:31 AM

Hi , Is there a way to send notification to airflow if the databricks job is success or failed, instead of calling databricks api multiple times to check status

Yuya Ebihara

05/08/2023, 9:45 AM

I have a question about

storageType

in deletion vectors. When I tried this feature in Databricks, it always return

. When do they use other types (

)? cc: @Marius Grama @Vikash Kumar @Slawomir Pajak

Slackbot

05/08/2023, 11:33 AM

This message was deleted.

Lucas Zago

05/08/2023, 3:03 PM

Hi guys what's the difference between

mergeSchema

and

overwriteSchema

? When to use one instead other?

Shreyas Venkat

05/08/2023, 3:55 PM

So I've been trying to go through the docs of delta-rs for python, I'm trying to connect a databricks workspace so I can access it locally without spark, I have the access token and the workspace URL in the ENV variables. But the issue I have is that in the example it says "data_catalog = DataCatalog.UNITY" and my interpreter just doesn't like that at all because it doesn't exist as an attribute. And I tried looking up the list of atributes on the website: https://delta-io.github.io/delta-rs/python/api_reference.html?highlight=datacatalog#deltalake.data_catalog.DataCatalog and I guess the list of data catalogs don't exist?

Kashyap Bhatt

05/08/2023, 4:19 PM

Add support for custom format to pyspark using python I want to read data from an on-prem Oracle schema (from multiple joined tables) as a Spark stream. E.g. How to create a custom streaming data source? Question is: Is it possible to write this in Python? Or does it have to be Java? Is Scala an option too? As I understand I need to create a new implementation of DataSourceV2 that will read from Oracle and store check point information (so I can invoke it at a flexible schedule and it'll know where to resume stream from) etc. and make my pyspark code, which uses the new format, look clean like:

Copy code

streaming_oracle_df = spark.readStream \
    .format("custom_oracle") \                 # <- my custom format
    .option("oracle_jdbc_str", "jdbc:...") \
    .option("custom_option1", 123) \
    .load()

streaming_oracle_df.writeStream \
    .trigger(availableNow=True) \
    .format('delta') \
    .option('checkpointLocation', '<s3://bucket/checkpoint/dim_customer>') \
    .start('<s3://bucket/tables/dim_customer>')

Jordan Cuevas

05/09/2023, 2:49 PM

Can anybody tell me if spark.databricks.delta.merge.enableLowShuffle is currently an option for open source delta lake? Broader question, is there a place to see all applicable config options? I've looked through the docs but haven't found it.

vijay

05/09/2023, 8:29 PM

Hi All, Can I use Mlflow to access delta tables within hive metastore? Looks like I need to provide delta table as one of the inputs to query it from within mlflow.

Artsiom Yudovin

05/09/2023, 9:04 PM

Hi All, I have faced a case when I have a vast parquet file inside Delta after the merge operation. It looks like Delta makes one huge file instead of making several files. Does anybody meet a similar case before? I use Apache Spark Structured Streaming with DeltaLake and merge each batch to the delta table.

Sumanth Bo3

05/10/2023, 6:35 AM

Hi All just started to explore delta lake and it’s capabilities recently it looks very interesting and promising --- My question is below is how i currently query my delta table

Copy code

df = spark.read.format("delta").load("/Users/saiteja/Downloads/new/")

df.createOrReplaceTempView("my_city_view")

result = spark.sql("SELECT LEFT(name, 1) AS letter, COUNT(*) AS num_cities FROM my_city_view GROUP BY letter ORDER BY letter ASC")

— is there any better efficient and faster way to query delta tables — Currently we are reading the entire delta table into df and creating a temporary view on top of it • Is the df and the tempView stored in memory during this phase • if the number of rows in the delta table is very high wouldn’t it cause high memory usage to read the table and create a view on top of it another thing which i observed is

SELECT LEFT(name, 1) AS letter, COUNT(*) AS num_cities FROM my_city_view GROUP BY letter ORDER BY letter ASC

my first query is relatively taking very high time to execute where are the subsequent queries on the same

my_city_view

are getting executed very fast here is an example

Copy code

No.of Records :150710
Time taken to run query 1: 135ms
Time taken to run query 2: 13ms
Time taken to run query 3: 15ms

is it caching the records in memory or something • i tried the same for different datasets but every time only the first query will take up more time where as queries executed after that will be 4 or 5 times faster Another question not related to delta lake — the delta lake query results in the above code is stored in

result

which is a spark dataframe when i try

result.show()

it is taking a good 40-50ms is the a better faster way to export the results mainly

Rahul Sharma

05/10/2023, 7:32 AM

hii Team , is there any way to change column data type in delta native table (my stack-holder are using column type int but i have boolean type they don’t want change the dashboard) please suggest me best way

Amit Panwar

05/10/2023, 2:26 PM

is there quick way to run delta lake on EMR

👀 1

John Darrington

05/10/2023, 9:12 PM

The concurrency control specified here - https://docs.delta.io/latest/concurrency-control.html - is really only if you're using Spark right? Not if, lets say, you're using the low level Rust library to interact directly?

Patrik Ekman

05/11/2023, 7:20 AM

Hello! I am wondering how you guys have handled delta streaming + eventual schema updates (changing column data types). Has your solution been to simply delete checkpoints/create a new checkpoint mirroring the new schema, or have you found some other nice workaround? It's in the classic bronze-silver-gold Azure platform, with delta streaming happening between all the zones. Thank you!

Alfonso

05/11/2023, 9:34 AM

Hello all!

👋 2

Alfonso

05/11/2023, 9:35 AM

One question, its there a way to add a generated column logic to an already created column? the resulting column is an already existing partition column with a trunc(col, “month”) format

Rudhra Raveendran

05/11/2023, 6:43 PM

Hi folks, does anyone know how to drop columns from a Delta Lake schema when the data is stored in Azure Blob Storage? E.g. I have data in storage account like so: abfss://container@storageaccount/path/to/data. path/to/data contains a bunch of folders in yyyy/mm/dd format, as well as a _delta_log folder. I'm trying to interact with this delta schema via scala in Azure synapse, e.g. val deltaTable = DeltaTable.forPath("abfss://..."). But then from there I'm not sure what to do, as all docs I can find mention writing SQL like ALTER TABLE table DROP COLUMNS columns, but I see no way to do that with this deltaTable. If I try spark.sql("SHOW TABLES") just to see what's there, I get an error about a null path, and if I try spark.sql("SELECT * FROM delta.`abfss://...`") that errors out too. Is there just something simple I'm missing? I feel like dropping columns should be a simple matter but I can't find any way to do it

Sukumar Nataraj

05/12/2023, 5:04 AM

Team, In production, when am doing the merge for a one delta table, kind of below log keeps printing and merge never ends. Can some one help me with this issue?. Delta 2.0.2 and Spark 3.2.1 , EMR 6.7.0, Can someone help me to debug this one. this something burning in production. Appreciate your help.

Copy code

23/05/12 05:01:16 INFO DataSourceStrategy: Pruning directories with: isnotnull(site_id#36183L),(site_id#36183L = 1007443),dynamicpruning#142232 [site_id#36183L]
23/05/12 05:01:16 INFO FileSourceStrategy: Pushed Filters: IsNull(deleted),IsNotNull(id)
23/05/12 05:01:16 INFO FileSourceStrategy: Post-Scan Filters: isnull(deleted#36235),isnotnull(id#36182L)
23/05/12 05:01:16 INFO FileSourceStrategy: Output Data Schema: struct<id: bigint, code: string, deleted: string ... 1 more fields>
23/05/12 05:01:16 INFO DataSourceStrategy: Pruning directories with: isnotnull(site_id#36173L),dynamicpruning#142231 [site_id#36173L]
23/05/12 05:01:16 INFO FileSourceStrategy: Pushed Filters: IsNotNull(id)
23/05/12 05:01:16 INFO FileSourceStrategy: Post-Scan Filters: isnotnull(id#36172L)
23/05/12 05:01:16 INFO FileSourceStrategy: Output Data Schema: struct<id: bigint>

Rahul Sharma

05/12/2023, 6:04 AM

Hii Team , is there any way to change the column schema in Delta Native table

Amit Panwar

05/12/2023, 12:07 PM

hi .. i dont want to use databrick , is it possible to use delta with EMR ?

Amit Panwar

05/12/2023, 12:07 PM

with all the feature

Amit Panwar

05/12/2023, 12:07 PM

Copy code

builder = SparkSession.builder.appName("MyApp") \
    .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
    .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog")

spark = configure_spark_with_delta_pip(builder).getOrCreate()
data = spark.range(0, 5)
deltaTable = spark.read.format("delta").load("/tmep/delta-table2")
deltaTable.show()