https://linen.dev logo
Join Slack
Powered by
This message was deleted.
# troubleshooting
s
This message was deleted.
s
what's the source? Do you have many files?
Also, the recommendation for 
druid.indexer.fork.property.druid.processing.numThreads
is 1 or 2. 10 seems high, but I doubt that is the issue.
Actually, aren't you using Indexers? I don't think the "fork" properties apply, but I could be wrong. Instead you control the total processing thread pool with 
druid.processing.numThreads
which should be set to the same value as the 
druid.worker.capacity
which should be set to (CPU cores - 1) if the Indexer runs on its own pod.
Still, none of that explains the fact that you get a single subtask, if you only have one file in your input source, that does explain it.
y
Yeah, I'm using the indexer
No, I have multiple files
All parquet file
s
Can you share your ingestion spec?
y
Copy code
{
  "type": "index_parallel",
  "spec": {
    "dataSchema": {
      "dataSource": "test",
      "timestampSpec": {
        "column": "timestamp_hourly",
        "format": "posix",
        "name": "__time"
      },
      "dimensionsSpec": {
        "dimensions": [
          ...
        ]
      },
      "metricsSpec": [ { "type" : "arrayOfDoublesSketch", ...},{ "type" : "longSum"...},{"type":"count"...}],
      "granularitySpec": {
        "type": "uniform",
        "segmentGranularity": "hour",
        "queryGranularity": "hour",
        "rollup": true
      }
    },
    "ioConfig": {
      "type": "index_parallel",
      "inputSource": {
        "type": "s3",
        "prefixes": ["s3://...."]
      },
      "inputFormat": {
        "type": "parquet"
      },
      "appendToExisting": false
    },
    "tuningConfig": {
      "type": "index_parallel",
      "maxNumConcurrentSubTasks":4,
      "maxRowsPerSegment": 20000,
      "maxRowsInMemory": 20000
    }
  }
}
s
What does the "index_parallel" task log say, it should start with some decision around how many tasks it will use. I'm assuming that the prefixes include the s3 path(s) that contain the parquet files. One thing that pops out at me is that there is no "partitionsSpec" which governs the parallel task behavior. I'm not sure if that affects it this way though. Since you are doing rollup, a partitionSpec makes sense to get perfect rollup, Range partitioning is probably best to get even segments sizes and pruning on the partitioning dimension(s). Also, have you considered using SQL based ingestion to do this, it uses a different, faster and more robust processing algorithm. You can start one of those directly from the console Query view, by using "Connect external data" and specify your S3 input source there and then just adjust the generated SQL to do the rollup by just coding it as a SQL aggregation query. You control the parallelism of the ingestion directly in the UI or in the query context:
y
Oh, I didn't even know that existed!!!! I will check it out now! Thanks
Damn it! Found the issue. We have many files but not big enough. Druid decided that it can just read them all in 1 task. Which makes sense! Added the following to test my theory
Copy code
"splitHintSpec": {
        "maxSplitSize": 1000000,
        "type": "maxSize"
      },
s
let us know how that goes. I was reading that as a max for each task before splitting into another task. But I didn't realize that it would also be used to control the number of tasks used. It's a good clarification for the docs.
y
Yeap, it does control how many tasks by looking at the file size. tested and verified. 🙂
s
I'll adjust the docs. Thanks!
👍 1
Hi @Younes Naguib, I think there is likely more nuance here. I ran a few tests with Druid 25.0 on a nano-quickstart laptop deployment which I adjusted to have a middle manager with 4 task slots: testing 2 small files, maxNumConcurrentSubtasks = 2
Copy code
dynamic, no splinthint  - 1 index_parallel task, 2 single_phase_sub_task
hashed, no splithint - 1 index_parallel, 2 partial_dimension_cardinality, 2 partial_index_generate, 1 partial_index_generic_merge
I did not have to change the split hint to get parallel tasks, I did need two input files. testing 2 small files, maxNumConcurrentSubtasks = 3
Copy code
hashed, no splithint - 1 index_parallel, 2 partial_dimension_cardinality, 2 partial_index_generate, 1 partial_index_generic_merge
In this case it was limited by the number of input files, as expected. testing 3 small files, maxNumConcurrentSubtasks = 3
Copy code
dynamic, no splithint - 1 index_parallel, 3 single_phase_sub_task
hashed, no splithint - 1 index_parallel, 3 partial_dimension_cardinality, 3 partial_index_generate, 1 partial_index_generic_merge
Verified by adding another file and I see 3 tasks running in the input processing phases for either dynamic or hashed partitioning I specified the list of files as a list of individual URIs, not sure if this has an effect. I'll try testing with like you did S3 prefixes instead.
👍 1
So, yes, when using S3 prefixes in the input source definition, it does not automatically generate more tasks unless the set of files it finds, exceed the splitHint specified (by default 1000 files or 1GiB total bytes).
👍 1
y
You rock 🙂
2 Views
Open in Slack
PreviousNext
Powered by