topic Re: When delta is a streaming source, how can we get the consumer lag? in Data Engineering

When delta is a streaming source, how can we get the consumer lag?

YFL — Thu, 09 Dec 2021 20:45:43 GMT

Hi,

I want to keep track of the streaming lag from the source table, which is a delta table.

I see that in query progress logs, there is some information about the last version and the last file in the version for the end offset, but this don't give the lag from the source table, unless I query it and check what the last version and files count is.

"sources" : [ {
    "description" : "DeltaSource[dbfs:/mnt/defaultDatalake/zones/bronze/my_source_table]",
    "startOffset" : {
      "sourceVersion" : 1,
      "reservoirId" : "15059b8a-0f48-4561-9424-8fcb0c8906de",
      "reservoirVersion" : 39673,
      "index" : -1,
      "isStartingVersion" : false
    },
    "endOffset" : {
      "sourceVersion" : 1,
      "reservoirId" : "15059b8a-0f48-4561-9424-8fcb0c8906de",
      "reservoirVersion" : 39674,
      "index" : -1,
      "isStartingVersion" : false
    },

Just to be clear, by lag I mean, that if for example the source table has the last row 100 and the streaming is now processing row 90, my lag would be 10 from the source table.

One more technical point: how can I parse the startOffset and endOffset. From the `SourceProgress` class I have direct access to the endOffset field, but not to its inners fields (like index). Should I just parse the endOffset string as json using some standard json library like jackson or ujson?

Thank you very much.

Re: When delta is a streaming source, how can we get the consumer lag?

YFL — Fri, 10 Dec 2021 08:12:16 GMT

Thanks, Kaniz. This is a highly important question for some production jobs we have (and we are highly invested in Databricks and Delta). I have seen others through the internet asking the same question, as well.

Thank you.

Yerachmiel Feltzman | Data Platform Developer

Re: When delta is a streaming source, how can we get the consumer lag?

jose_gonzalez — Fri, 10 Dec 2021 23:05:33 GMT

Hi @Yerachmiel Feltzman

You can take a look at the following metrics https://docs.databricks.com/delta/delta-streaming.html#metrics in your stream query progress

Re: When delta is a streaming source, how can we get the consumer lag?

YFL — Sun, 12 Dec 2021 09:09:45 GMT

Hi, @Jose Gonzalez , I don't see in the link something that states the lag from the source delta table.

Thanks anyway.

Re: When delta is a streaming source, how can we get the consumer lag?

jose_gonzalez — Mon, 13 Dec 2021 17:24:14 GMT

Hi @Yerachmiel Feltzman

Are you able to see these metrics?

{

"sources" : [

{

"description" : "DeltaSource[file:/path/to/source]",

"metrics" : {

"numBytesOutstanding" : "3456",

"numFilesOutstanding" : "8"

}

]

}

Re: When delta is a streaming source, how can we get the consumer lag?

YFL — Mon, 13 Dec 2021 17:51:32 GMT

I am.

Yerachmiel Feltzman | Data Platform Developer

Re: When delta is a streaming source, how can we get the consumer lag?

Anonymous — Tue, 28 Dec 2021 16:15:56 GMT

@Yerachmiel Feltzman - Does the fact you can see the metrics resolve the issue?

Re: When delta is a streaming source, how can we get the consumer lag?

YFL — Mon, 03 Jan 2022 09:33:32 GMT

Hi, @Piper Wilson . Those metrics don't solve the issue. What I am asking for is to keep tracking the lag between the running streaming and the source delta table. Those metrics give me a lot of information, like input rates (as bytes and as files) and some other metrics, but lag from the source table I don't see there.

Again, by lag, I mean "the difference between the source table last record position/timestamp and the current row my streaming is processing". For example (simplistic), the source table has 100 rows and the streaming has processed 90, it is lagging 10 rows.

What I have in mind is something similar to Kafka's consumer lag, but another approach is welcome as well. The whole point here is: "How do I keep track and know where my streaming is regarding its source? How can I know if it is processing slow than expected, ie, slower than its source delta table?"

Thanks. 🙂

Re: When delta is a streaming source, how can we get the consumer lag?

jose_gonzalez — Wed, 26 Jan 2022 00:43:52 GMT

Hi @Yerachmiel Feltzman ,

You will need to take a look at the micro-batch metrics. This article will explain more what each metric means https://databricks.com/blog/2020/07/29/a-look-at-the-new-structured-streaming-ui-in-apache-spark-3-0.html

Re: When delta is a streaming source, how can we get the consumer lag?

Anonymous — Thu, 12 May 2022 13:44:35 GMT

Hey @Yerachmiel Feltzman

I hope all is well.

Just wanted to check in if you were able to resolve your issue or do you need more help? We'd love to hear from you.

Thanks!

Re: When delta is a streaming source, how can we get the consumer lag?

YFL — Sun, 15 May 2022 12:53:30 GMT

Hey @Vartika Nain . The issue was resolved. Thanks.

Re: When delta is a streaming source, how can we get the consumer lag?

chinhvu1111 — Sun, 18 Aug 2024 18:43:33 GMT

Hello, How did you solve this problem?
Could you kindly share it with me? I have the same problem.

I also want to check more details.