Spark Jobs

const cluster = await client.getSparkCluster({ name: "emr-prod" });
console.log(cluster.type, cluster.properties);

const { scripts, hasMore } = await client.listSparkJobs();

let offset = 0;
const allScripts: string[] = [];
while (true) {
  const page = await client.listSparkJobs({ limit: 50, offset });
  allScripts.push(...page.scripts);
  if (!page.hasMore) break;
  offset += 50;
}

const { runId } = await client.submitSparkJob({
  namespace: "my-namespace",
  name: "daily-etl",
  entrypoint: "etl_pipeline.py",
  args: ["--date", "2026-03-11"],
  executorNumbers: 4,
});

const run = await client.getRun(runId);

const { runId } = await client.submitSparkJob({
  cluster: "emr-prod",
  namespace: "finance",
  name: "daily-etl",
  entrypoint: "s3://my-bucket/jobs/etl_pipeline.py",
  args: ["--date", "2026-03-11"],
  pyFiles: "s3://my-bucket/jobs/deps.zip",
  packages: ["org.example:my-lib:1.0.0"],
});

const { runId, state, run } = await client.submitSparkJobAndWait({
  namespace: "my-namespace",
  name: "daily-etl",
  entrypoint: "etl_pipeline.py",
  pollIntervalMs: 5000,
  timeoutMs: 300000,
});

if (state === "COMPLETE") {
  const elapsed = run.duration;
  // proceed with downstream work ...
} else {
  throw new Error(`Run ${runId} ended with state: ${state}`);
}

const run = await client.getRun(runId);

if (run.state === "COMPLETE") {
  const duration = run.duration;
  const jobName = run.job.name;
  // handle completion ...
} else if (run.state === "FAIL") {
  const error = run.error;
  // handle failure ...
}