https://linen.dev logo
Join Slack
Powered by
This message was deleted.
# general
s
This message was deleted.
j
Did you set rollup= true on this table? If so, then possibly your "rolled up" data is still fragmented, so the query is completing the rollup process on the fly. This is standard behavior in Druid, it could account for the difference. Fyi if this is the case, then when you run a compaction on your datasource it should clean up the discrepancy by completing the data rollup process.
m
It’s not the rollup datasource. It’s the raw datasource.
Let me clarify - We had some data already in the data source, we need to perform some correction activity. So we marked all the datasource segments as unused and all the segments got dropped.
after dropping all the segments we are ingesting the data again from the object store in a batch ingestion format.
j
Your segment sizes are very tiny, so you have lots of segments. I would suggest run a quick compaction ... using MSQ SQL this is as simple as:
Copy code
replace into <datasource> overwrite all
select * from <datasource>
partitioned by <DAY|MONTH>
I am thinking the compaction will clean up this discrepancy.
m
Copy code
Error: Plan validation failed

Cannot execute REPLACE with SQL engine 'native'.

org.apache.calcite.tools.ValidationException
Getting this execption when I execute the above query in druid query console
You are suggesting to peform the compaction and then execute the sql count command?
Copy code
{{
        "type": "compact",
        "dataSource": "{datasource}",
        "ioConfig": {{
            "type": "compact",
            "inputSpec": {{
                "type": "interval",
                "interval": "{start_date}/{end_date}"
            }}
        }},
        "tuningConfig": {{
            "type": "index_parallel",
            "maxRowsPerSegment": 5000000,
            "logParseExceptions": true
        }},
        "context": {{
            "storeCompactionState": true,
            "priority": 25,
            "forceTimeChunkLock": true
        }}
    }}
Here is my compaction script.
Is anything required to make changes in the compaction script? or it looks ok?
Compaction is not solving..
Do we need to reset the metadata of the datasource?
@John Kowtko
I think if we reset the metadata then we might loos the Kafka partition offsets right? but after the compaction, we are not able to get the right value. Still the aggregation count and datasource row counts are mismatching.
Or is the above compaction script not correct?
Also, FYI, We are not performing the SQL-based ingestion, We are performing the native batch ingestion.
Also, if we create a new data source then we don’t encounter this issue. Only if we drop the segments and re ingest using the native batch format then we are encountering this issue.
j
Did you get a new (and smaller) set of segments after the compaction?
For MSQ, you probably need to enable the extension in order for it to show up and be selected as the engine to use for a "replace" statement.
Another thing to try -- copy the current datasource into a new one, and see if the new one has the correct counts in it.
a
can you run this query?
Copy code
select count(*) from table
m
Yea, select count is giving invalid records count. It’s not matching with the row count
What I have noticed is if we drop all the segments and reingest all the data using native batch ingestion format then we are encountering it.
The one compaction script which I have posted here is correct? Something needs to be updated to perform again?
druid-multi-stage-query
Is this the extension i have configure to work MSQ?
Do we have any other workaround other than this MSQ? Since we have to add this extension in the production and make the deployment.
j
yes, 
druid-multi-stage-query
is the correct extension.
m
And one more question, All our datasource segment granularity is “Day” and only one segment is available per day. How does compaction again will going to help?
j
I only mentioned using MSQ because the syntax is simpler than native JSON spec ... you can still do all of this in native, the spec is just a bit harder to write. Your native spec above shows double braces ... not sure if that will work as-is, but yes that looks like a standard compaction spec to me.
m
Ok..
We have written some Python script to execute the compaction, I just copied that snippet from the source code, thats why you are seeing the double braces.
Is this something a bug within the druid? or is it some weird issue? do I need to restart or refresh the data sources using any APIs.?
a
How was the data originally ingested? Was it streaming ingestion based? Could you also please share if any of the segments are overshadowed in this datasource from the segments tab?
m
Yes, It’s a kafka stream based ingestion.
a
I see. Did you suspend the streaming ingestion and wait for the publish and handoff before marking the segments as unused?
m
Yes, I suspened the stream based ingestion and marked the segments as unused.
a
I see. Also, it appears that the partition number has reached 32768 which is the max limit for segment ids using dynamic partitioning
Are you seeing any errors with segment allocation?
m
I don’t see any errors. The native batch ingestion and kafka based ingestion are getting success.
but when we query the data i am getting a empty result.
Segments and datasources everything available but when we query we are getting empty results
FYI - I have opened one of the segment records and copied the identifier property value and queried using that property but i am getting the empty result.
a
One thing you can also do is get a json query corresponding to this SQL query. You can do that through 
explain
dialogue. In the json query, if you add 
bySegment
to 
true
in the query context, you will get per-segment results.
m
Ok. I will try, but why it’s not working if we just normally query on the native batch ingested data.
a
John had pointed out that you shard specs are of the type numbered_overwrite which are created only when using segment locking
Was it intentional to use segment locking instead of timechunk locking?
m
I think We didn’t mention any locking configuration. Mostly, it should have taken the default values.
a
Your compaction tasks have 
forceTimeChunkLock
set to true
m
Yes..
Copy code
"context": {
        "storeCompactionState": true,
        "priority": 25,
        "forceTimeChunkLock": true,
         "taskLockTimeout": 1000000   

    }
a
It should also be set to true for your kafka and index_parallel tasks
m
Our batch ingestion is index_parallel. all the data has ingested with native batch ingestion with “index_parallell” task.
a
Yes, that is the task type. Could you please share the context from the task spec?
m
Ok..
Copy code
"context": {
    "forceTimeChunkLock": false,
    "useLineageBasedSegmentAllocation": true
  }
a
Ah, could you please retry the ingestion with forceTimeChunkLock set to true for all your tasks? (Which version of druid is this btw?)
m
It’s 24.0.1
How do we correct the ingested data? Instead of ingesting again? Is this possible?
a
Are you appending or replacing data with the index_parallel jobs?
m
append…
Copy code
"appendToExisting": true,
      "dropExisting": false
a
We found that the issue was related to https://github.com/apache/druid/pull/13070 and the use of segment locking
5 Views
Open in Slack
PreviousNext
Powered by