Scheduled batch supervisor by abhishekrb19 · Pull Request #17353 · apache/druid

abhishekrb19 · 2024-10-15T17:43:03Z

This change introduces a scheduled batch supervisor in Druid. The supervisor periodically wakes up to submit an MSQ ingest query, allowing users to automate batch ingestion directly within Druid. Think of it as simple batch task workflows natively integrated into Druid, though it doesn't replace more sophisticated workflow management systems like Apache Airflow. This is an experimental feature.

Summary of changes:

The scheduled_batch supervisor can be configured as follows:

{
    "type": "scheduled_batch",
    "schedulerConfig": {
        "type": "unix",
        "schedule": "*/5 * * * *"
    },
    "spec": {
        "query": "REPLACE INTO foo OVERWRITE ALL SELECT * FROM bar PARTITIONED BY DAY"
    },
    "suspended": false
}

The supervisor will submit the REPLACE sql query repeatedly every 5 minutes. The supervisor supports two types of cron scheduler configurations:

Unix cron syntax:
- Type must be set to unix.
- Follows the standard Unix cron format, e.g., */5 * * * * to schedule the SQL task every 5 minutes.
- Supports macro expressions such as @daily, @hourly, @monthly, etc.
Quartz cron syntax:
- Type must be set to quartz.
- Offers more flexibility and control for scheduling tasks.
- Example: 0 0 0 ? 3,6,9,12 MON-FRI to schedule tasks at midnight on weekdays during March, June, September, and December.

Key points:

User can specify the query along with any context in the spec. This structure is identical to what the MSQ task engine accepts.
Currently, the batch supervisor will repeatedly submit the DML SQL present in the spec as-is on its schedule.
This can be useful for recurring batch ingestion or reindexing tasks.
There is no parameterization or "change detection" support for input sources yet, but some simple mechanisms are planned for the future.
Users can configure multiple scheduled batch supervisors for the same datasource.
Only DML ingestion queries are currently supported by the scheduled batch supervisor.
Users can suspend, resume and terminate the scheduled batch supervisor. Suspending the supervisor stops new tasks from being scheduled.

Some implementation details:

The scheduled batch supervisor runs on the Overlord and communicates with the Broker to:
a. Validate and parse the user-specified query.
b. Submit MSQ queries to the /druid/v2/sql/task/ endpoint.
A ScheduledBatchScheduler is injected in the Overlord, which is responsible for scheduling and unscheduling all scheduled batch instances.
A BrokerClient implementation has been added, leveraging the ServiceClient functionality.
The SqlTaskStatus and its unit test SqlTaskStatusTest have been moved from the msq module to the sql module so it can be reused by the BrokerClient implementation in the sql module.
Added ExplainPlanInformation class, which is used to deserialize the explain plan response from the Broker.

The status API response for the supervisor contains the scheduler state along with active and completed tasks:

curl -X GET http://localhost:8888/druid/indexer/v1/supervisor/scheduled_batch__foo__3a916bcf-060d-4a42-aab6-904ff61d25cd/status

Details

{
  "supervisorId": "scheduled_batch__foo__3764d545-62b8-4d52-a20c-568b009268e1",
  "state": "RUNNING",
  "lastTaskSubmittedTime": "2025-02-11T01:39:00.119Z",
  "nextTaskSubmissionTime": "2025-02-11T01:40:00.000Z",
  "taskReport": {
    "totalSubmittedTasks": 5,
    "totalSuccessfulTasks": 3,
    "totalFailedTasks": 2,
    "recentTasks": [
      {
        "taskStatus": {
          "id": "query-bb0fc147-2cde-4935-95ed-bf40a26568de",
          "status": "SUCCESS",
          "duration": 26628,
          "errorMsg": null,
          "location": {
            "host": null,
            "port": -1,
            "tlsPort": -1
          }
        },
        "updatedTime": "2025-02-11T01:37:26.780Z"
      },
      {
        "taskStatus": {
          "id": "query-1d51d1fc-ece5-4376-b64d-716ccecbd708",
          "status": "SUCCESS",
          "duration": 26367,
          "errorMsg": null,
          "location": {
            "host": null,
            "port": -1,
            "tlsPort": -1
          }
        },
        "updatedTime": "2025-02-11T01:38:26.493Z"
      },
      {
        "taskStatus": {
          "id": "query-baa8d243-ff57-4c18-b513-865687dd3e93",
          "status": "FAILED",
          "duration": 26197,
          "errorMsg": "Task execution process exited unsuccessfully with code[2]. See middleManager logs for more details.",
          "location": {
            "host": null,
            "port": -1,
            "tlsPort": -1
          }
        },
        "updatedTime": "2025-02-11T01:39:26.355Z"
      }
    ]
  }
}

Future Improvements:

Scheduler State Persistence: This feature doesn't yet persist the scheduler state. Adding persistence will improve robustness (e.g., preventing missed jobs during Overlord restarts).
Task Limits: There is currently no limit on the maximum number of tasks allowed by the scheduled batch supervisor. We may want to consider this in the future.
The Supervisors page in the Druid web-console should change a bit to accommodate this change (more broadly non-streaming supervisors).

Release Note

This change introduces an experimental feature called scheduled batch supervisor in Druid. The supervisor periodically wakes up to submit an MSQ ingest query, allowing users to automate batch ingestion directly within Druid. Think of it as simple batch task workflows natively integrated into Druid, though it doesn't replace more sophisticated workflow management systems like Apache Airflow. Currently, the supervisor will repeatedly submit the same MSQ query as-is. The payload for the supervisor is below, please check the PR summary for additional details:

{
    "type": "scheduled_batch",
    "schedulerConfig": {
        "type": "unix",
        "schedule": "*/5 * * * *"
    },
    "spec": {
        "query": "REPLACE INTO foo OVERWRITE ALL SELECT * FROM bar PARTITIONED BY DAY"
    },
    "suspended": false
}

This PR has:

…ation.

Please enter the commit message for your changes. Lines starting

… reusability (#17542) This PR contains non-functional / refactoring changes of the following classes in the sql module: 1. Move ExplainPlan and ExplainAttributes fromsql/src/main/java/org/apache/druid/sql/http to processing/src/main/java/org/apache/druid/query/explain 2. Move sql/src/main/java/org/apache/druid/sql/SqlTaskStatus.java -> processing/src/main/java/org/apache/druid/query/http/SqlTaskStatus.java 3. Add a new class processing/src/main/java/org/apache/druid/query/http/ClientSqlQuery.java that is effectively a thin POJO version of SqlQuery in the sql module but without any of the Calcite functionality and business logic. 4. Move BrokerClient, BrokerClientImpl and Broker classes from sql/src/main/java/org/apache/druid/sql/client to server/src/main/java/org/apache/druid/client/broker. 5. Remove BrokerServiceModule that provided the BrokerClient. The functionality is now contained in ServiceClientModule in the server package itself which provides all the clients as well. This is done so that we can reuse the said classes in #17353 without brining in Calcite and other dependencies to the Overlord.

…batch_supervisor

The jobsExecutor is reponsible for submitting jobs to the broker for all scheduled batch supervisors. The cronExecutor was simply decoupling the submitting of jobs from the actual scheduled running of jobs. The service client library already has async threads, so we remove the cronExecutor to keep things simple as things are handled in an async manner by the service client. If we ever observe and see evidence of bottlenecks around task submission, etc, we should be able to make jobsExecutor multiple threaded instead of single threaded.

+    // do nothing
+  }
+
+  public enum State implements SupervisorStateManager.State


kfaraz

Sorry for the delay on this, @abhishekrb19 !

I have tried to do a complete review. Once we have addressed these comments, we should be good to merge this PR.

Thanks a lot for your patience!! 🙂

abhishekrb19 · 2025-02-10T09:00:09Z

@kfaraz, thank you for the reviews! I believe I have addressed and/or responded to all of your comments.

kfaraz

Minor non-blocking comments. 🚀

kfaraz · 2025-02-11T03:01:43Z

+      @JsonProperty("taskStatus") TaskStatus taskStatus,
+      @JsonProperty("updatedTime") DateTime updatedTime


Nit: Since this constructor is not used for deserializing right now, we don't need the @JsonProperty annotation.

kfaraz · 2025-02-11T03:03:36Z

+  }
+
+  /**
+   * Used for internal tracking. So this field is *not* Jackson serialized to avoid


Nit: I am not sure if markdown is honored in javadocs. Use <i>, <b> or caps instead for emphasis.

kfaraz · 2025-02-11T03:29:10Z

@abhishekrb19 , the changes look good to me. You may proceed with the merge.

When you get the time though, please take another look at this comment
#17353 (comment)
If you feel that it's valid, it can be addressed in a follow up PR.

abhishekrb19 · 2025-02-11T06:50:28Z

Sounds good, just responded in the comment inline

Initial version of scheduled batch supervisor implementation.

2afb4a9

github-actions Bot added Area - Batch Ingestion Area - Querying Area - Dependencies Area - Ingestion Area - MSQ For multi stage queries - https://github.com/apache/druid/issues/12262 labels Oct 15, 2024

abhishekrb19 added 2 commits October 15, 2024 12:57

Minor logging adjustments.

3a20d0c

Address first round of static check issues.

7d4b93a

abhishekrb19 marked this pull request as draft October 15, 2024 21:49

abhishekrb19 added 2 commits October 15, 2024 17:54

Merge branch 'master' into scheduled_batch_supervisor

80df85d

Forbidden API check and checkstyle.

5dd0e63

github-advanced-security AI found potential problems Oct 16, 2024

View reviewed changes

abhishekrb19 added 5 commits October 15, 2024 20:26

Fix checkstyle and tests with new brokerClient.

a15b0f0

Class renames and fixup tests.

e18b183

Move SqlTaskStatusTest to sql package as well alongside its implement…

c6b79d9

…ation.

Rename tests to align with impl.

179f8fa

Adjust test

9d25c9a

github-advanced-security AI found potential problems Oct 16, 2024

View reviewed changes

Comment thread indexing-service/src/main/java/org/apache/druid/indexing/batch/ScheduledBatchSupervisor.java Fixed

Comment thread ...xing-service/src/main/java/org/apache/druid/indexing/batch/ScheduledBatchSupervisorSpec.java Fixed

abhishekrb19 added 12 commits October 15, 2024 22:25

Intellij inspections.

11b6135

Fix

8a0d660

BrokerServiceModule.

23d73de

Try licenses.yaml

5a33da6

Remove unused dependency and add missing undeclared dep.

a4aa243

Simplify explain plan stuff.

cdcdf59

Please enter the commit message for your changes. Lines starting

Merge branch 'master' into scheduled_batch_supervisor

00d133c

Better unix and quartz scheduler mapping and tests.

21bfc75

Only unix cron supports macros.

45fbd58

Remove extraneous changes.

dfed561

Merge branch 'master' into scheduled_batch_supervisor

38f2447

Lines.

0ef4a6d

Merge branch 'master' into scheduled_batch_supervisor

6b57055

abhishekrb19 added 11 commits December 6, 2024 10:06

Merge branch 'master' into scheduled_batch_supervisor

b5b459e

Merge branch 'master' into scheduled_batch_supervisor_with_broker

740b5be

Merge branch 'scheduled_batch_supervisor_with_broker' into scheduled_…

8ed2854

…batch_supervisor

Merge branch 'master' into scheduled_batch_supervisor

fb4fd92

Merge branch 'master' into scheduled_batch_supervisor

4894011

A shared single-threaded jobs executor.

07a0094

Suspended supervisor stops further scheduling.

8c17ea8

Checkstyle.

76e94b3

Remove unused supervisor state enum values.

f10296b

Remove avatica dependency.

5403a0f

github-advanced-security AI found potential problems Dec 20, 2024

View reviewed changes

Verify state again inside the runnable.

5addd13

kfaraz reviewed Feb 5, 2025

View reviewed changes

abhishekrb19 added 4 commits February 9, 2025 22:20

Merge branch 'master' into scheduled_batch_supervisor

debfadb

Review

edfa8da

Remove transitive dep licenses

14ff95b

Add back jakarta license.

389ef26

kfaraz reviewed Feb 10, 2025

View reviewed changes

Review

619fd0d

kfaraz approved these changes Feb 11, 2025

View reviewed changes

abhishekrb19 merged commit f776f33 into apache:master Feb 11, 2025

kgyrtkirk added this to the 33.0.0 milestone Apr 14, 2025

clintropolis mentioned this pull request Jun 18, 2025

Web console: improve SQL autocomplete and add JSON autocomplete #18126

Merged

soenkeliebau mentioned this pull request Jul 24, 2025

Compile list of new product features in newly supported versions for the 25.7.0 release (for the blog post) stackabletech/issues#750

Closed

		@JsonProperty("taskStatus") TaskStatus taskStatus,
		@JsonProperty("updatedTime") DateTime updatedTime

Conversation

abhishekrb19 commented Oct 15, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Summary of changes:

Release Note

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Check notice

Uh oh!

kfaraz left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

abhishekrb19 commented Feb 10, 2025

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

kfaraz left a comment

Choose a reason for hiding this comment

Uh oh!

kfaraz Feb 11, 2025

Choose a reason for hiding this comment

Uh oh!

kfaraz Feb 11, 2025

Choose a reason for hiding this comment

Uh oh!

kfaraz commented Feb 11, 2025

Uh oh!

abhishekrb19 commented Feb 11, 2025

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

abhishekrb19 commented Oct 15, 2024 •

edited

Loading