What are the use cases to use Delta.io?

I have a problem trying to run self-made function using the TRANSFORM() in Databricks. Does someone know the solution?

CREATE TEMPORARY FUNCTION 
increment_one(x DOUBLE) RETURNS DOUBLE RETURN x + 1;

SELECT 
  typeof(my.nested.data.factor), -- array<array<double>>
  TRANSFORM(my.nested.data, x -> transform(x, y -> y.factor + 1)) as this_works,
  -- Output: [[2], [2]]
  TRANSFORM(my.nested.data, x -> transform(x, y -> abs(y.factor))) as also_works,
  -- Output: [[1], [1]]
  TRANSFORM(my.nested.data, x -> transform(x, y -> increment_one(y.factor))) as does_not_work
  -- Output: AnalysisException
FROM my_dummy_data

The exception thrown is:

AnalysisException: Resolved attribute(s) y#4454 missing from  in operator !Project [cast(lambda y#4454.factor as double) AS x#4455].; line 7 pos 59

I would expect that I could call the

increment_one()

similarly as

abs()

. I would expect them both to return a double. If I replace the

y.factor

with a literal value such as 5 (thus calling

increment_one(5)

inside the nested transform), my function will work as expected.

Hi All, I have worked on Spark extensively for 6 years now starting from Hortonworks, MapR and Cloudera. I have brief understanding of running the spark jobs on EMR. When I am moving on Databricks Cloud? How and what I can learn? I have gone through below videos, is it enough? These are from Databricks. https://vimeo.com/735869688 https://vimeo.com/735869485 https://vimeo.com/491450691 https://vimeo.com/742641085 https://vimeo.com/735871756 https://vimeo.com/676044575 Thank you

The increasing demand for data with low latency at Coinbase has required them to renovate their data ingestion and replication framework. SOON (Spark cOntinuOus iNgestion) is designed to ingest the incremental changes (appends, updates, and deletes) to an existing table on Delta Lake. Take a look at the overall architecture for SOON and learn how Coinbase leverages Delta Lake’s change data feed (CDF) feature. ✅ SOON for near real-time data at Coinbase - Part 1 ➡️ https://lnkd.in/eqn-t_D8

Hi fellows, my question is not about delta lake but generally for Spark on EMR - sorry about that. So, if it does not fit here, please feel free to delete the question. I have a Spark job which I am running on EMR. This job has to process some 2 TB of data and I am getting error:

# java.lang.OutOfMemoryError: Java heap space
# -XX:OnOutOfMemoryError="kill -9 %p"
#   Executing /bin/sh -c "kill -9 27571"...

I have tried using various instance types like 20-25 core instances of m5.16xlarge and r5.12xlarge. Also, tried playing around spark configurations like spark.driver.memory and spark.executor.memory from 30g to 300g, but nothing helped. The job has not any major computation but simply its spark.coalesce.write.partitionBy.parquet. Also tried setting _HADOOP_HEAPSIZE_ in the configuration to 100g and 200g. Please ask if more information is required. Thanks. Here is the screenshot of EMR executors:

Placing this random LinkedIn plug in the #random channel 🤪 Delta Lake has officially surpassed 19K followers on LinkedIn. WOOHOO! This means we have under 1,000 followers to go to reach 20k. Help us reach this milestone by following Delta Lake on LinkedIn - PLUS, we have some really awesome content to help you learn more about Delta Lake's features, community, events, & more. Thanks all! 😄🦀 https://www.linkedin.com/company/deltalake

Anyone else having issues accessing their databricks workspace on AWS? Getting a blank page on 3 different browsers across 2 machines.

Have you heard the news? Delta Lake users can now use Conda natively to manage their delta-spark dependency! 🦀 Try it out today ➡️ https://anaconda.org/conda-forge/delta-spark

Hi All, 1. Is there anyway to stop the Databricks workflow job using pyspark or python? 2. How to find the running jobs in one workflow and how to get the run id for the same using python? can anyone plz help me on this? Thanks

Hello 👋 I have a question regarding exporting delta tables from one databricks workspace to another. I have a heavy volume data that I processed in lower environment, lets say dev. Now, can I have an option to just export the ingested data into prod and then run incremental jobs on prod to reduce the time/cost/effort? Posted in #deltalake-questions

Hi All, I am trying to create a Databricks table on top of my sql table using JDBC connection *using sql only syntax. I*t works when the credentials are hardcoded, but fails when credentials stored in azure key vault is being used for connection.

a.usr= dbutils.secrets.get(scope = "test1",key = "test1-key",)

a.pwd = dbutils.secrets.get(scope = "test1",key = "test2-key"

,)

CREATE TEMPORARY VIEW test

USING JDBC

options(driver=${<http://a.td|a.td>_driver},

url=${<http://a.td|a.td>_url},

dbtable=${<http://a.td|a.td>_tbl},

user=${a.usr},

password=${a.pwd}

)

How to format/handle the a.usr & a.pwd from extracted from azure key vault inside the JDBC options parameters in the sql way. Py way of parameterizing the credentials works, but want to do it the sql way to minimize lot of changes existing app. Thanks !

Hi All, Is there any way to do ‘dos2unix’ cmd in Databricks notebook. My usecase is I have a csv file in ADLS location but with some junk chars so I need to remove the junk chars while reading the file using pyspark in my databricks notebook. I can handle this using dos2unix cmd running manually but this suppose to be automate. Can anyone plz help on this? Thanks!

Can someone please help me to connect data lake to databricks

Anyone using Databricks/delta lake but use Glue catalog tables instead of metastore or unity catalog? thoughts on your experience?

Hi all, how's everyone doing? We have an issue with our metastore slowing down read and write to our delta tables. Topic: Databricks Metastore Cluster Details: Databricks runtime 10.4 ML LTS Apache Spark 3.2.1, Scala 2.12 We have initially attempted to communicate with databricks support to understand the Metastore is Down Issue. We were told it's due to BoneCP trying to established failed connections from the connection pool etc. Attempt 1: The resolution was to try using HikariCP instead of BoneCP, after adding HikariCP datanucleas jar files to /databricks/hive.

spark.hadoop.datanucleus.connectionPoolingType hikari

We got the same error as below. Attempt 2: Docs : • https://learn.microsoft.com/en-us/azure/databricks/release-notes/runtime/10.4ml • https://learn.microsoft.com/en-us/azure/databricks/release-notes/runtime/10.4#hikaricp-is-now-the-default-hive-metastore-connection-pool installed packages: x So 10.4 ML is different to 10.4 in that 10.4 ML is not using hikari (and has the below libraries missing from the databricks Libraries UI) • We installed the following (because this is the standard from 10.4 non-ml version):

org.datanucleus:datanucleus-api-jdo:4.2.4
org.datanucleus:datanucleus-core:4.1.17
org.datanucleus:datanucleus-rdbms:4.1.19
org.datanucleus:javax.jdo:3.2.0-m3

configuration that we've tried:

spark.hadoop.datanucleus.connectionPoolingType hikari

spark.hadoop.datanucleus.connectionPoolingType HikariCP

spark.databricks.hive.metastore.client.pool.type hikari

spark.databricks.hive.metastore.client.pool.type HikariCP

All jar files were added and we are getting the following error after restarting. ``org.datanucleus.api.jdo.JDOPersistenceManagerFactory.freezeConfiguration(JDOPersistenceManagerFactory.java:775)`

... 128 more

`Caused by: java.lang.Throwable: Attempt to invoke the "hikari" plugin to create a ConnectionPool gave an error : The connection pool plugin of type "hikari" was not found in the CLASSPATH!`` Any one with knowledge on how to resolve this issue, we would appreciate your help. Thanks

Hi all, someone know a good tutorial about how to integrate unit_tests on databricks ?

Does anyone by any chance know of a slack/chat where spark/pyspark/jdbc developers hang around? I'm playing around with a small framework and something seems to have changed from databricks LTS 10.4 to LTS 11.3 that causes reading with pyspark-jdbc-sqlserver to generate SQL ending in a invalid 'WHERE (1)' when there should not be any where clause at all as far as I can tell. It seems to be caused by joining/comparing incoming dataframe from sqlserver to a empty delta table before writing. Once the table has some data the exact same query runs fine.

Hi All, How to handle ‘concurrency exception’ while trying to merge data into delta table using pyspark? Can anyone plz help on this?

x-posting here

The deltalake 0.7.0 release includes exciting new features including a cleaner API for file listing operations, the ability to manually create checkpoints, and the power to fetch all the add actions of a Delta table! This post by @Will Jones & @Matthew Powers demonstrates the new features and explains when you can use them in your production workflows. 🦀 https://lnkd.in/e7_PqEGi

Hi, not exactly a delta question but was hoping to maybe get some input here (couldn’t find reading material about it). I have the following 2 options (as an example), they should be reading the same exact data and writing the same exact data, though one transforms the data outside of the batch and then writes it using the

foreachBatch

method, and the other transform and writes inside the batch function. Are their side effects/issues/things worth taking into consideration? Thanks! Example:

Hi All, I need your help on below scenario. Kindly help me to clear on this case. In Databricks notebook, I am consuming data from Kafka to delta lake using Pyspark. If suppose I want to know the consumer lag , consumer group details means how I can find these details using pyspark or atleast do we have any other way to do it? My Databricks run time version: Databricks 7.5 (includes Apache Spark 3.0.1, Scala 2.12) Thanks in advance!

Does anyone happen to know any spark tricks that can be used to influence on what workers/in memory partitions the scan/read 'stages' put induvidual records? Even though neither data sources/data frames have hard partitions per se, it would be nice to be able to co-locate based on key columns on read to avoid exchange/shuffle etc later on when doing joins or using window functions. I suppose the command repartition(partitionExprs : Column)* is the closest but even that happens after the data is read if I'm understanding correctly.

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 Everyone, I am facing issue in retry mechanism for DBR >=11.3. The job is running fine with DBR 10.4 and retry is also working as expected. But in DBR 11.3, if any error comes, the spark is unable to initialize or it does not laod everything in second attempt or more. Is anyone familiar with this..?

(Py)Spark community on Slack Hi, I really enjoy this kind of Delta Lake community on Slack that we have right here. I was wondering if there is something similar for (Py)Spark users. Perhaps someone has a tip for me? The official page states that there is a Mailing List and Stack Overflow but I had not much luck addressing my questions there - and I like the communication over slack more. 😄

Hello, I'm trying to create a JDBC connection to my local spark environment. I'm looking to "*com.simba.spark.jdbc4.Driver"* JAR file - Version 4. Someone has de jar file with this driver? Thanks

Hi All, I am getting below error when trying to get the value of timestamp field. Can anyone plz help me on this to fix it? Elastic Search data in my index- index1: { “id”: “123”, “__ts”: 1649289672103 } Pyspark code: es_df = spark.read.format(“es”).options(**es_conn_details) .options(“es.read.metadata”,”true”) .options(“es.read.metadata.version”,”true”) .options(“es.mapping.date.rich”,”false”) .load(“index1”) .select(“id”, “__ts”) es_xfm_df = es_df.selectExpr(“id”, “from_unixtime(__ts/1000, ‘yyyy-MM-dd HHmmss’) as __ts”) Error when try to execute display(es_xfm_df): RuntimeException: Error while encoding: Java.lang.RuntimeException: Java.lang.Long is not a valid external type for schema of string Caused by: RuntimeException: Java.lang.Long is not a valid external type for schema of string Thanks in Advance

Hello, I have nearly three years of experience in setting up and managing Databricks (Lakehouse) in AWS, based on Delta Lake of course, and I am looking into expanding my skillset to Azure Databricks too. If anyone has any suggestions of online tutorials/books/courses/resources that have proven to be a high quality, please share. Ideal topics would be around "Azure Databricks for those who know AWS" and also "Azure for those who know AWS".

Hi, I am following this document to verify multi cluster writes on Amazon EMR. https://delta.io/blog/2022-05-18-multi-cluster-writes-to-delta-lake-storage-in-s3/ I do not see either the table or record getting created in DynamoDB. Am I missing anything? Any inputs are much appreciated. Below are my configurations

SparkSession spark = SparkSession.builder()
        .config("fs.s3.impl", "org.apache.hadoop.fs.s3a.S3AFileSystem")
        .config("fs.AbstractFileSystem.s3.impl", "org.apache.hadoop.fs.s3a.S3A")
        .config("fs.s3a.aws.credentials.provider", "com.amazonaws.auth.DefaultAWSCredentialsProviderChain")
        .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")
        .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog")
        .config("spark.delta.logStore.s3.impl", "io.delta.storage.S3DynamoDBLogStore")
        .config("spark.io.delta.storage.S3DynamoDBLogStore.ddb.tableName", "delta_log")
        .config("spark.io.delta.storage.S3DynamoDBLogStore.ddb.region", "us-east-1")
        .config("spark.io.delta.storage.S3DynamoDBLogStore.credentials.provider",
            "com.amazonaws.auth.DefaultAWSCredentialsProviderChain")
        .getOrCreate();