Skip to content

Spark instrumentation: capture emr step id when spark runs as EMR step #10670

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
gh-worker-dd-mergequeue-cf854d merged 13 commits into from
Feb 26, 2026
Merged

Spark instrumentation: capture emr step id when spark runs as EMR step #10670

Show file tree
Hide file tree
Changes from 7 commits
Commits
Show all changes
13 commits
Select commit Hold shift + click to select a range
2d6992e
make span attr for emr step id captured from workdir
aboitreaud Feb 23, 2026
1a62eb6
capture step id for launcher spans
aboitreaud Feb 23, 2026
369fce0
step id extraction error resilience
aboitreaud Feb 24, 2026
594e9cf
add class as helper
aboitreaud Feb 24, 2026
bbf7b20
debug log
aboitreaud Feb 24, 2026
95b82cc
rename to EmrUtils
aboitreaud Feb 24, 2026
3aea275
Add unit tests
aboitreaud Feb 24, 2026
ee7c12e
Gate emr_step_id attribute to emr-detected runtimes
aboitreaud Feb 25, 2026
cfa491c
spotless apply
aboitreaud Feb 25, 2026
a8fdf19
switch to keys contain .emr. check
aboitreaud Feb 25, 2026
d3c7ebb
remove isRunningOnEMR
aboitreaud Feb 26, 2026
34af8f9
early returns to avoid deep nesting
aboitreaud Feb 26, 2026
e0bec00
Merge branch 'master' into adrien.boitreaud/capture-emr-step-id
aboitreaud Feb 26, 2026
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
8 changes: 8 additions & 0 deletions ...ommon/src/main/java/datadog/trace/instrumentation/spark/AbstractDatadogSparkListener.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -276,6 +276,7 @@ private void initApplicationSpanIfNotInitialized() {
}

captureApplicationParameters(builder);
captureEmrStepId(builder);

Optional<OpenlineageParentContext> openlineageParentContext =
OpenlineageParentContext.from(sparkConf);
Expand Down Expand Up @@ -1210,6 +1211,13 @@ private void captureApplicationParameters(AgentTracer.SpanBuilder builder) {
builder.withTag("config.spark_version", sparkVersion);
}

private static void captureEmrStepId(AgentTracer.SpanBuilder builder) {
String stepId = EmrUtils.getEmrStepId();
if (stepId != null) {
builder.withTag("emr_step_id", stepId);
}
}

private void captureJobParameters(AgentTracer.SpanBuilder builder, Properties properties) {
for (Tuple2<String, String> conf : sparkConf.getAll()) {
if (SparkConfAllowList.canCaptureJobParameter(conf._1)) {
Expand Down
38 changes: 38 additions & 0 deletions ...tation/spark/spark-common/src/main/java/datadog/trace/instrumentation/spark/EmrUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,38 @@
package datadog.trace.instrumentation.spark;

import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import javax.annotation.Nullable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/** Extracts the AWS EMR Step ID from the working directory name (e.g. s-07767992IY7VC5NVV854). */
class EmrUtils {

private static final Logger log = LoggerFactory.getLogger(EmrUtils.class);
private static final Pattern EMR_STEP_ID_PATTERN = Pattern.compile("^(s-[0-9A-Za-z]+)$");

@Nullable
static String getEmrStepId() {
try {
String userDir = System.getProperty("user.dir");
if (userDir != null) {
Path workDir = Paths.get(userDir).getFileName();
if (workDir != null) {
Matcher matcher = EMR_STEP_ID_PATTERN.matcher(workDir.toString());
if (matcher.matches()) {
log.debug("EMR step ID extracted: {}", matcher.group(1));
return matcher.group(1);
Copy link
Copy Markdown
Contributor

@pawel-big-lebowski pawel-big-lebowski Feb 24, 2026

Choose a reason for hiding this comment

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

Please consider creating unit test for this method. This would be really helpful in understanding what paths match the regex.

Also, is there any way to figure out if the launcher is run on EMR? Otherwise, it's hard to say if this aint adding emr tags to non-emr jobs.

Copy link
Copy Markdown
Contributor Author

@aboitreaud aboitreaud Feb 24, 2026

Choose a reason for hiding this comment

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

Added unit tests in EmrUtilsTest.java !

The regex is specific enough to only match the EMR step ids, so I think we are guaranteed to only add this tag on Spark on EMR spans.

Copy link
Copy Markdown
Contributor

@pawel-big-lebowski pawel-big-lebowski Feb 25, 2026

Choose a reason for hiding this comment

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

The regex matches all work dirs starting with s-. Are you saying this is good enough to tell Spark is running on EMR?

}
}
}
} catch (Throwable t) {
log.debug("Unable to extract EMR step ID from working directory", t);
}
return null;
}

private EmrUtils() {}
}
4 changes: 3 additions & 1 deletion ...ommon/src/main/java/datadog/trace/instrumentation/spark/SparkLauncherInstrumentation.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,7 +32,9 @@ public String instrumentedType() {
@Override
public String[] helperClassNames() {
return new String[] {
packageName + ".SparkConfAllowList", packageName + ".SparkLauncherListener",
packageName + ".EmrUtils",
packageName + ".SparkConfAllowList",
packageName + ".SparkLauncherListener",
};
}

Expand Down
8 changes: 8 additions & 0 deletions ...spark-common/src/main/java/datadog/trace/instrumentation/spark/SparkLauncherListener.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,7 @@ public static synchronized void createLauncherSpan(Object launcher) {
.start();
span.setSamplingPriority(PrioritySampling.USER_KEEP, SamplingMechanism.DATA_JOBS);
setLauncherConfigTags(span, launcher);
captureEmrStepId(span);
launcherSpan = span;

if (!shutdownHookRegistered) {
Expand Down Expand Up @@ -131,6 +132,13 @@ public void infoChanged(SparkAppHandle handle) {
}
}

private static void captureEmrStepId(AgentSpan span) {
String stepId = EmrUtils.getEmrStepId();
if (stepId != null) {
span.setTag("emr_step_id", stepId);
}
Copy link
Copy Markdown
Contributor

@pawel-big-lebowski pawel-big-lebowski Feb 24, 2026

Choose a reason for hiding this comment

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

can we also add a tag this is EMR launcher? does it make sense?

Copy link
Copy Markdown
Contributor Author

@aboitreaud aboitreaud Feb 24, 2026 •
edited
Loading

Choose a reason for hiding this comment

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

spark.application are tagged them with step id, above in the other Listener.

regarding marking these as emr launcher, I think the span operation name spark.launcher.launch already tells us this is a launcher + seeing step_id means we're necessarily on EMR, so I'd say we don't need more than that but let me know how you feel

}

/**
* Extract launcher configuration via reflection and set as span tags. Secret redaction uses the
* default pattern only (not spark.redaction.regex) because the SparkLauncher conf map is a plain
Expand Down
41 changes: 41 additions & 0 deletions ...on/spark/spark-common/src/test/java/datadog/trace/instrumentation/spark/EmrUtilsTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,41 @@
package datadog.trace.instrumentation.spark;

import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNull;

import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;

class EmrUtilsTest {

private String originalUserDir;

@BeforeEach
void saveUserDir() {
originalUserDir = System.getProperty("user.dir");
}

@AfterEach
void restoreUserDir() {
System.setProperty("user.dir", originalUserDir);
}

@Test
void returnsStepIdWhenWorkdirMatchesEmrPattern() {
System.setProperty("user.dir", "/mnt/var/lib/hadoop/steps/s-07767992IY7VC5NVV854");
assertEquals("s-07767992IY7VC5NVV854", EmrUtils.getEmrStepId());
}

@Test
void returnsNullWhenWorkdirDoesNotMatchEmrPattern() {
System.setProperty("user.dir", "/home/hadoop");
assertNull(EmrUtils.getEmrStepId());
}

@Test
void returnsNullForApplicationIdWorkdir() {
System.setProperty("user.dir", "/home/hadoop/application_1234567890_0001");
assertNull(EmrUtils.getEmrStepId());
}
}
1 change: 1 addition & 0 deletions ...spark_2.12/src/main/java/datadog/trace/instrumentation/spark/Spark212Instrumentation.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@ public String[] helperClassNames() {
packageName + ".AbstractSparkPlanSerializer",
packageName + ".AbstractSparkPlanUtils",
packageName + ".DatabricksParentContext",
packageName + ".EmrUtils",
packageName + ".OpenlineageParentContext",
packageName + ".DatadogSpark212Listener",
packageName + ".PredeterminedTraceIdContext",
Expand Down
1 change: 1 addition & 0 deletions ...spark_2.13/src/main/java/datadog/trace/instrumentation/spark/Spark213Instrumentation.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@ public String[] helperClassNames() {
packageName + ".AbstractSparkPlanSerializer",
packageName + ".AbstractSparkPlanUtils",
packageName + ".DatabricksParentContext",
packageName + ".EmrUtils",
packageName + ".OpenlineageParentContext",
packageName + ".DatadogSpark213Listener",
packageName + ".PredeterminedTraceIdContext",
Expand Down