Create Dask Delta writer

[dask-contrib]

Create Dask Delta writer

[rajagurunath]

Hi @MrPowers,

I really like the idea of leveraging delta-like metadata handling for parquet files also.

Currently, Dask seems to collect footer information from every parquet file using Pyarrow (read_metadata) utilities and creates a global _metadata file.

So if we are tracking that information in a JSON file, What information we are planning to save in the JSON logs? and how we are going to collect that information?

I gave an initial read to this Delta protocol, Are we planning to save just the statistics, schema information for every parquet file or some more things? and also how we are planning to make use of this information in dask readers perspective.

I am still confused here, so can you please shed some more light on this!

Thanks

[MrPowers]

@rajagurunath - the global _metadata file that Dask is currently creating is not scalable and I think we should ignore that file for purposes of this project. Assume write_metadata_file=False.

Let's look at a little example and see what the outputted JSON file should look like:

df = pd.DataFrame(
    {"letter": ["a", "b", "c", "a", "a", "d"], "number": [1, 2, 3, 4, 5, 6]}
)
ddf = dd.from_pandas(df, npartitions=3)
ddf.to_delta("tmp/parquet/1", engine="pyarrow", write_metadata_file=False)

Here's the JSON file that should be outputted:

{
   "commitInfo":{
      "timestamp":1632491414394,
      "operation":"WRITE",
      "operationParameters":{
         "mode":"Overwrite",
         "partitionBy":"[]"
      },
      "isBlindAppend":false,
      "operationMetrics":{
         "numFiles":"3",
         "numOutputBytes":"2390",
         "numOutputRows":"6"
      }
   }
}{
   "protocol":{
      "minReaderVersion":1,
      "minWriterVersion":2
   }
}{
   "metaData":{
      "id":"db102a08-5265-4f86-a281-dfc8cccacf0e",
      "format":{
         "provider":"parquet",
         "options":{
            
         }
      },
      "schemaString":"{\"type\":\"struct\",\"fields\":[{\"name\":\"letter\",\"type\":\"string\",\"nullable\":true,\"metadata\":{}},{\"name\":\"number\",\"type\":\"long\",\"nullable\":true,\"metadata\":{}}]}",
      "partitionColumns":[
         
      ],
      "configuration":{
         
      },
      "createdTime":1632491409910
   }
}{
   "add":{
      "path":"part.0.parquet",
      "partitionValues":{
         
      },
      "size":377,
      "modificationTime":1632491410000,
      "dataChange":true
   }
}{
   "add":{
      "path":"part.1.parquet",
      "partitionValues":{
         
      },
      "size":674,
      "modificationTime":1632491410000,
      "dataChange":true
   }
}{
   "add":{
      "path":"part.2.parquet",
      "partitionValues":{
         
      },
      "size":654,
      "modificationTime":1632491410000,
      "dataChange":true
   }
}

Here's what the files should look like:

tmp/parquet/1/
  _delta_log/
    00000000000000000000.json
  part.0.parquet
  part.1.parquet
  part.2.parquet

We basically need to create the _delta_log exactly how Spark creates it when it writes Delta files. See this file for a code snippet on how to create Delta lakes with Spark.

Let me know if you have any additional questions. You're doing great work and I'm happy to help!

[mrocklin]

🎉

…

On Mon, Jul 10, 2023 at 9:46 AM Florian Jetter ***@***.***> wrote: Closed #4 <#4> as completed via #29 <#29>. — Reply to this email directly, view it on GitHub <#4 (comment)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AACKZTHFCQ4UXNJJBDIUE6LXPQISTANCNFSM5E6IUIRQ> . You are receiving this because you are subscribed to this thread.Message ID: ***@***.*** com>