You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@druid.apache.org by kf...@apache.org on 2025/02/28 03:55:17 UTC
(druid) branch master updated: Support pushing and streaming task payload for HDFS (#17742)
This is an automated email from the ASF dual-hosted git repository.
kfaraz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/druid.git
The following commit(s) were added to refs/heads/master by this push:
new f15ba0884ae Support pushing and streaming task payload for HDFS (#17742)
f15ba0884ae is described below
commit f15ba0884ae7d7d01f20b30fa4a6445408b11719
Author: Virushade <70...@users.noreply.github.com>
AuthorDate: Fri Feb 28 11:55:11 2025 +0800
Support pushing and streaming task payload for HDFS (#17742)
Implement pushTaskPayload/streamTaskPayload as introduced in #14887
for HDFS storage to allow larger mm-less ingestion payloads when using
HDFS as the deep storage location.
---
distribution/docker/peon.sh | 2 +-
.../druid/storage/hdfs/tasklog/HdfsTaskLogs.java | 20 +++++++++++++++++
.../indexing/common/tasklogs/HdfsTaskLogsTest.java | 25 ++++++++++++++++------
3 files changed, 40 insertions(+), 7 deletions(-)
diff --git a/distribution/docker/peon.sh b/distribution/docker/peon.sh
index 8103f475ccb..3b4dfc4326b 100755
--- a/distribution/docker/peon.sh
+++ b/distribution/docker/peon.sh
@@ -161,4 +161,4 @@ fi
# If TASK_JSON is not set, CliPeon will pull the task.json file from deep storage.
mkdir -p ${TASK_DIR}; [ -n "$TASK_JSON" ] && echo ${TASK_JSON} | base64 -d | gzip -d > ${TASK_DIR}/task.json;
-exec bin/run-java ${JAVA_OPTS} -cp $COMMON_CONF_DIR:$SERVICE_CONF_DIR:lib/*: org.apache.druid.cli.Main internal peon $@
+exec bin/run-java ${JAVA_OPTS} -cp $COMMON_CONF_DIR:$SERVICE_CONF_DIR:lib/*: org.apache.druid.cli.Main internal peon --taskId ${TASK_ID} $@
diff --git a/extensions-core/hdfs-storage/src/main/java/org/apache/druid/storage/hdfs/tasklog/HdfsTaskLogs.java b/extensions-core/hdfs-storage/src/main/java/org/apache/druid/storage/hdfs/tasklog/HdfsTaskLogs.java
index f89b0fafde9..535598bc2d7 100644
--- a/extensions-core/hdfs-storage/src/main/java/org/apache/druid/storage/hdfs/tasklog/HdfsTaskLogs.java
+++ b/extensions-core/hdfs-storage/src/main/java/org/apache/druid/storage/hdfs/tasklog/HdfsTaskLogs.java
@@ -207,6 +207,26 @@ public class HdfsTaskLogs implements TaskLogs
}
}
}
+
+ @Override
+ public void pushTaskPayload(String taskId, File taskPayloadFile) throws IOException
+ {
+ final Path path = getTaskPayloadFileFromId(taskId);
+ log.info("Pushing payload for task[%s] to location[%s]", taskId, path);
+ pushTaskFile(path, taskPayloadFile);
+ }
+
+ @Override
+ public Optional<InputStream> streamTaskPayload(String taskId) throws IOException
+ {
+ final Path path = getTaskPayloadFileFromId(taskId);
+ return streamTaskFile(path, 0);
+ }
+
+ private Path getTaskPayloadFileFromId(String taskId)
+ {
+ return new Path(mergePaths(config.getDirectory(), taskId.replace(':', '_') + ".payload.json"));
+ }
}
diff --git a/extensions-core/hdfs-storage/src/test/java/org/apache/druid/indexing/common/tasklogs/HdfsTaskLogsTest.java b/extensions-core/hdfs-storage/src/test/java/org/apache/druid/indexing/common/tasklogs/HdfsTaskLogsTest.java
index 9d0273d1a95..125fbd28872 100644
--- a/extensions-core/hdfs-storage/src/test/java/org/apache/druid/indexing/common/tasklogs/HdfsTaskLogsTest.java
+++ b/extensions-core/hdfs-storage/src/test/java/org/apache/druid/indexing/common/tasklogs/HdfsTaskLogsTest.java
@@ -50,7 +50,7 @@ public class HdfsTaskLogsTest
final File tmpDir = tempFolder.newFolder();
final File logDir = new File(tmpDir, "logs");
final File logFile = new File(tmpDir, "log");
- Files.write("blah", logFile, StandardCharsets.UTF_8);
+ Files.asCharSink(logFile, StandardCharsets.UTF_8).write("blah");
final TaskLogs taskLogs = new HdfsTaskLogs(new HdfsTaskLogsConfig(logDir.toString()), new Configuration());
taskLogs.pushTaskLog("foo", logFile);
@@ -69,11 +69,11 @@ public class HdfsTaskLogsTest
final File logFile = new File(tmpDir, "log");
final TaskLogs taskLogs = new HdfsTaskLogs(new HdfsTaskLogsConfig(logDir.toString()), new Configuration());
- Files.write("blah", logFile, StandardCharsets.UTF_8);
+ Files.asCharSink(logFile, StandardCharsets.UTF_8).write("blah");
taskLogs.pushTaskLog("foo", logFile);
Assert.assertEquals("blah", readLog(taskLogs, "foo", 0));
- Files.write("blah blah", logFile, StandardCharsets.UTF_8);
+ Files.asCharSink(logFile, StandardCharsets.UTF_8).write("blah blah");
taskLogs.pushTaskLog("foo", logFile);
Assert.assertEquals("blah blah", readLog(taskLogs, "foo", 0));
}
@@ -87,7 +87,7 @@ public class HdfsTaskLogsTest
final TaskLogs taskLogs = new HdfsTaskLogs(new HdfsTaskLogsConfig(logDir.toString()), new Configuration());
- Files.write("{}", statusFile, StandardCharsets.UTF_8);
+ Files.asCharSink(statusFile, StandardCharsets.UTF_8).write("{}");
taskLogs.pushTaskStatus("id", statusFile);
Assert.assertEquals(
"{}",
@@ -95,6 +95,19 @@ public class HdfsTaskLogsTest
);
}
+ @Test
+ public void test_taskPayload() throws Exception
+ {
+ final File tmpDir = tempFolder.newFolder();
+ final File logDir = new File(tmpDir, "logs");
+ final File payload = new File(tmpDir, "payload.json");
+ final TaskLogs taskLogs = new HdfsTaskLogs(new HdfsTaskLogsConfig(logDir.toString()), new Configuration());
+
+ Files.asCharSink(payload, StandardCharsets.UTF_8).write("{}");
+ taskLogs.pushTaskPayload("id", payload);
+ Assert.assertEquals("{}", StringUtils.fromUtf8(ByteStreams.toByteArray(taskLogs.streamTaskPayload("id").get())));
+ }
+
@Test
public void testKill() throws Exception
{
@@ -107,7 +120,7 @@ public class HdfsTaskLogsTest
final TaskLogs taskLogs = new HdfsTaskLogs(new HdfsTaskLogsConfig(logDir.toString()), new Configuration());
- Files.write("log1content", logFile, StandardCharsets.UTF_8);
+ Files.asCharSink(logFile, StandardCharsets.UTF_8).write("log1content");
taskLogs.pushTaskLog("log1", logFile);
Assert.assertEquals("log1content", readLog(taskLogs, "log1", 0));
@@ -118,7 +131,7 @@ public class HdfsTaskLogsTest
long time = (System.currentTimeMillis() / 1000) * 1000;
Assert.assertTrue(fs.getFileStatus(new Path(logDirPath, "log1")).getModificationTime() < time);
- Files.write("log2content", logFile, StandardCharsets.UTF_8);
+ Files.asCharSink(logFile, StandardCharsets.UTF_8).write("log2content");
taskLogs.pushTaskLog("log2", logFile);
Assert.assertEquals("log2content", readLog(taskLogs, "log2", 0));
Assert.assertTrue(fs.getFileStatus(new Path(logDirPath, "log2")).getModificationTime() >= time);
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org