Cfrgtkky

I'm trying out Structured Streaming in Azure Databricks using Databricks storage dbfs:/ as checkpoint and file storage locations. I'm streaming event data from an Azure Event Hub to parquet files on dbfs.

When I cancel the query and restart I get the dreaded illegal state exception.

Caused by: java.lang.IllegalStateException: dbfs:/test02/_spark_metadata/2 doesn't exist when compacting batch 9 (compactInterval: 10)

I'm trying to figure out the relationship between the Azure Event hubs conf startingPosition option, the checkpointing offset and commit logs and the storage location _spark_metadata information.

Why do I get an illegal state on restart and how are the different logs and metadata information used by Spark?

There are 10 offset log files named 0 through 9. 9 commit log files named 0 through 8. Obviously there ought to be another commit log file if the last batch had been processed completely.

The commmit log files seem to contain no pertinent information why I guess it's their pure existence (name) that is important.

commit log file:

v1

{"nextBatchWatermarkMs":0}

The offset logs only contain information about the offsets per partition.

{

    "batchWatermarkMs": 0,

    "batchTimestampMs": 1542634173589,

    "conf": {

        "spark.sql.shuffle.partitions": "200",

        "spark.sql.streaming.stateStore.providerClass": "org.apache.spark.sql.execution.streaming.state.HDFSBackedStateStoreProvider",

        "spark.sql.streaming.multipleWatermarkPolicy": "min",

        "spark.sql.streaming.flatMapGroupsWithState.stateFormatVersion": "2"

    }

}

{

    "poc-test01": {

        "1": 3666,

        "0": 3664

    }

}

If I understood correctly Structured Streaming has a cycle where it first writes a Write Ahead Log in the offset/ folder. Then process the batch data. Then write a Commit Log in the commit/ folder. Each batch triggers a new cycle.

The _spark_metadata/ folder only contain two log files, 0 and 1, each file contains information about two parquet files in the filepath/ folder.

{

    "path": "dbfs:/test02/part-00000-98c5def2-bdf3-4e56-af9f-ff0796344b01-c000.snappy.parquet",

    "size": 59180,

    "isDir": false,

    "modificationTime": 1542214122000,

    "blockReplication": 1,

    "blockSize": 536870912,

    "action": "add"

}

{

    "path": "dbfs:/test02/part-00001-a38dbf8b-240a-48d0-8438-f81f0506a79c-c000.snappy.parquet",

    "size": 59462,

    "isDir": false,

    "modificationTime": 1542214122000,

    "blockReplication": 1,

    "blockSize": 536870912,

    "action": "add"

}

There are 10 parquet files in the filepath/ folder, the event hub contain two partitions.

My best guess now is that the _spark_metadata/ folder is used for idempotence in the file sink. Spark ought to be able to track the processed batches from the source offset positions to the created files in the filepath/ folder wherein the _spark_metadata/ folder is located. Somehow this broke.

• Is the combination of checkpointing and file store metadata handling documented somewhere? It seems like a huge blackbox that seem to work so so.


• Is this a structured streaming bug or a file sink checkpoint bug?


• How can I go back to the last working state?