reading delta-lake

Question

reading delta-lake

arkiboys 9,706

hello,
in databricks I get the data as follows:

column_1 _year _month _day

----------------------------------------------------

xxx 2022 11 14
yyy 2022 11 14
aaa 2022 11 15
...
ttt 2022 11 16
...

I would like to get the data into partitions in delta-lake as follows

delta
_delta_log
_year=2022
_month=11
_day=14
_day=15
...

to do this, I am using the following pyspark in databricks
folder_path = "/merged/presentation/delta/"
df_present.write.mode("overwrite").format("delta").partitionBy('_year', '_month', '_day').save(f"/mnt/storage_dev/curated{folder_path}")

so then I can query the data-lake as follows:
display(spark.read.format("delta").load(f'/mnt/storage_dev/curated{aspect_cu_folder_path}'))

The problem I have is that the data is displayed for the latest _day whereas I want to see all data in _day=14 and _day=15 and ...
do you see what I am doing wrong in the pyspark?
thank you

MartinJaffer-MSFT 26,236 Reputation points

2022-11-17T22:48:59.537+00:00
Hi again @arkiboys

So you want to load data in a more sophisticated way than path/day=4. More like path/day=4 or 5, or maybe path/ where day > 3

It looks similar to pyspark-optimize-read-load-from-delta-using-selected-columns-or-partitions

From best-practices.html another looks like

val path = "..." val partition = "year = '2019'" val numFilesPerPartition = 16 spark.read .format("delta") .load(path) .where(partition)

other options leverage the table part of delta, and in the back-end use the where clause to select only relevant partitions to read.

From delta-batch.html

If you then run the following query:
spark.sql('SELECT * FROM default.events WHERE eventTime >= "2020-10-01 00:00:00" <= "2020-10-01 12:00:00"')
Delta Lake automatically generates a partition filter so that the preceding query only reads the data in partition year=2020/month=10/day=01 even if a partition filter is not specified.

You can use an EXPLAIN clause and check the provided plan to see whether Delta Lake automatically generates any partition filters.
MartinJaffer-MSFT 26,236 Reputation points

2022-11-22T18:05:46.817+00:00

Hello @arkiboys ,
We haven’t heard from you on the last response and was just checking back to see if you have a resolution yet. In case if you have any resolution please do share that same with the community as it can be helpful to others . Otherwise, will respond back with the more details and we will try to help .

Thanks
Martin
MartinJaffer-MSFT 26,236 Reputation points

2022-11-30T18:46:56.057+00:00

Hello @arkiboys ,
We haven’t heard from you on the last response and was just checking back to see if you have a resolution yet. In case if you have any resolution please do share that same with the community as it can be helpful to others .
If you have any question relating to the current thread, please do let us know and we will try out best to help you.
In case if you have any other question on a different issue, we request you to open a new thread .

Thanks
Martin

Accepted answer

0 additional answers

Your answer

MartinJaffer-MSFT 26,236 Reputation points

2022-11-17T22:48:59.537+00:00

Hi again @arkiboys

So you want to load data in a more sophisticated way than path/day=4. More like path/day=4 or 5, or maybe path/ where day > 3

It looks similar to pyspark-optimize-read-load-from-delta-using-selected-columns-or-partitions

From best-practices.html another looks like

val path = "..." val partition = "year = '2019'" val numFilesPerPartition = 16 spark.read .format("delta") .load(path) .where(partition)

other options leverage the table part of delta, and in the back-end use the where clause to select only relevant partitions to read.

From delta-batch.html

If you then run the following query:
spark.sql('SELECT * FROM default.events WHERE eventTime >= "2020-10-01 00:00:00" <= "2020-10-01 12:00:00"')
Delta Lake automatically generates a partition filter so that the preceding query only reads the data in partition year=2020/month=10/day=01 even if a partition filter is not specified.

You can use an EXPLAIN clause and check the provided plan to see whether Delta Lake automatically generates any partition filters.
MartinJaffer-MSFT 26,236 Reputation points

2022-11-22T18:05:46.817+00:00

Hello @arkiboys ,
We haven’t heard from you on the last response and was just checking back to see if you have a resolution yet. In case if you have any resolution please do share that same with the community as it can be helpful to others . Otherwise, will respond back with the more details and we will try to help .

Thanks
Martin
MartinJaffer-MSFT 26,236 Reputation points

2022-11-30T18:46:56.057+00:00

Hello @arkiboys ,
We haven’t heard from you on the last response and was just checking back to see if you have a resolution yet. In case if you have any resolution please do share that same with the community as it can be helpful to others .
If you have any question relating to the current thread, please do let us know and we will try out best to help you.
In case if you have any other question on a different issue, we request you to open a new thread .

Thanks
Martin

Answer 1

Hello @arkiboys ,
Thanks for the question and using MS Q&A platform.

As I understand you want to read from a range of partitions, not just a single partition.

I admit, delta is my weakness. From what I see, reading a single file like that is not necessary because of the table metadata. Due to "Partition discovery," we don't have to specify individual files, and can use a different method. All this is done for you if you access through the table rather than file directly. We can verify this using .explain at the end.

try

spark.sql("select * from tablename where year = '2022' and month = '11' and day > 13")

or

spark.read.table("tablename").where("year = '2022' and month = '11' and day > 13")

Excerpt from best-practices

Load a single partition: Reading partitions directly is not necessary. For example, you don’t need to run spark.read.format("parquet").load("/data/date=2017-01-01"). Instead, use a WHERE clause for data skipping, such as spark.read.table("<table_name>").where("date = '2017-01-01'").

Please do let me if you have any queries.

Thanks
Martin

Please don't forget to click on or upvote button whenever the information provided helps you. Original posters help the community find answers faster by identifying the correct answer. Here is how
Want a reminder to come back and check responses? Here is how to subscribe to a notification
- If you are interested in joining the VM program and help shape the future of Q&A: Here is how you can be part of Q&A Volunteer Moderators

Share via

reading delta-lake

0 additional answers

Your answer