Skip to content

Make IngestSegmentFirehoseFactory splittable for parallel ingestion #7048

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
jihoonson merged 7 commits into from
Apr 2, 2019
Merged

Make IngestSegmentFirehoseFactory splittable for parallel ingestion #7048

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
ba8a758
Make IngestSegmentFirehoseFactory splittable for parallel ingestion
glasser Feb 27, 2019
77c6b26
Code review feedback
glasser Mar 1, 2019
602af30
Add missing @JsonProperty
glasser Mar 1, 2019
87c65bb
Integration test passes
glasser Mar 8, 2019
abe27fb
Add unit test
glasser Mar 9, 2019
5842d50
Remove two FIXME comments from CompactionTask
glasser Mar 9, 2019
e7de623
Updates from code review
glasser Mar 22, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 3 additions & 1 deletion docs/content/ingestion/firehose.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -87,7 +87,8 @@ The below configurations can be optionally used for tuning the firehose performa
### IngestSegmentFirehose

This Firehose can be used to read the data from existing druid segments.
It can be used ingest existing druid segments using a new schema and change the name, dimensions, metrics, rollup, etc. of the segment.
It can be used to ingest existing druid segments using a new schema and change the name, dimensions, metrics, rollup, etc. of the segment.
This firehose is _splittable_ and can be used by [native parallel index tasks](./native_tasks.html#parallel-index-task).
Copy link
Copy Markdown
Contributor

@jihoonson jihoonson Mar 11, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Would you please add IngestSegmentFirehose to https://github.com/apache/incubator-druid/blob/master/docs/content/ingestion/native_tasks.md#parallel-index-task too?

A sample ingest firehose spec is shown below -

```json
Expand All @@ -106,6 +107,7 @@ A sample ingest firehose spec is shown below -
|dimensions|The list of dimensions to select. If left empty, no dimensions are returned. If left null or not defined, all dimensions are returned. |no|
|metrics|The list of metrics to select. If left empty, no metrics are returned. If left null or not defined, all metrics are selected.|no|
|filter| See [Filters](../querying/filters.html)|no|
|maxInputSegmentBytesPerTask|When used with the native parallel index task, the maximum number of bytes of input segments to process in a single task. If a single segment is larger than this number, it will be processed by itself in a single task (input segments are never split across tasks). Defaults to 150MB.|no|

#### SqlFirehose

Expand Down
2 changes: 1 addition & 1 deletion docs/content/ingestion/native_tasks.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,7 +45,7 @@ task statuses. If one of them fails, it retries the failed task until the retryi
If all worker tasks succeed, then it collects the reported list of generated segments and publishes those segments at once.

To use this task, the `firehose` in `ioConfig` should be _splittable_. If it's not, this task runs sequentially. The
current splittable fireshoses are [`LocalFirehose`](./firehose.html#localfirehose), [`HttpFirehose`](./firehose.html#httpfirehose)
current splittable fireshoses are [`LocalFirehose`](./firehose.html#localfirehose), [`IngestSegmentFirehose`](./firehose.html#ingestsegmentfirehose), [`HttpFirehose`](./firehose.html#httpfirehose)
, [`StaticS3Firehose`](../development/extensions-core/s3.html#statics3firehose), [`StaticAzureBlobStoreFirehose`](../development/extensions-contrib/azure.html#staticazureblobstorefirehose)
, [`StaticGoogleBlobStoreFirehose`](../development/extensions-contrib/google.html#staticgoogleblobstorefirehose), and [`StaticCloudFilesFirehose`](../development/extensions-contrib/cloudfiles.html#staticcloudfilesfirehose).

Expand Down
2 changes: 2 additions & 0 deletions indexing-service/src/main/java/org/apache/druid/indexing/common/task/CompactionTask.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -506,10 +506,12 @@ private static IndexIOConfig createIoConfig(
new IngestSegmentFirehoseFactory(
dataSchema.getDataSource(),
interval,
null,
null, // no filter
// set dimensions and metrics names to make sure that the generated dataSchema is used for the firehose
dataSchema.getParser().getParseSpec().getDimensionsSpec().getDimensionNames(),
Arrays.stream(dataSchema.getAggregators()).map(AggregatorFactory::getName).collect(Collectors.toList()),
null,
toolbox.getIndexIO(),
coordinatorClient,
segmentLoaderFactory,
Expand Down
Loading