1import * as glue from "@distilled.cloud/aws/glue";
2import type * as Duration from "effect/Duration";
3import * as Effect from "effect/Effect";
4import * as Stream from "effect/Stream";
5import { Unowned } from "../../AdoptPolicy.ts";
6import { isResolved } from "../../Diff.ts";
7import { createPhysicalName } from "../../PhysicalName.ts";
8import * as Provider from "../../Provider.ts";
9import { Resource } from "../../Resource.ts";
10import { createInternalTags, hasAlchemyTags } from "../../Tags.ts";
11import { toWireMinutes } from "../../Util/Duration.ts";
12import { AWSEnvironment } from "../Environment.ts";
13import type { Providers } from "../Providers.ts";
14import {
15 fetchObservedTags,
16 jobArn,
17 retryWhileRoleNotAssumable,
18 syncTags,
19} from "./internal.ts";
20
21export interface JobCommand {
22 /**
23 * The command name: `glueetl` (Spark), `pythonshell` (Python shell), or
24 * `gluestreaming` (Spark streaming).
25 */
26 name: "glueetl" | "pythonshell" | "gluestreaming" | (string & {});
27 /** The S3 path to the job script, e.g. `s3://my-bucket/scripts/job.py`. */
28 scriptLocation?: string;
29 /** For `pythonshell`, the Python version (`3` or `3.9`). */
30 pythonVersion?: string;
31 /** The Ray runtime version (for Ray jobs). */
32 runtime?: string;
33}
34
35export interface JobProps {
36 /**
37 * Name of the job. If omitted, a unique name is generated. Changing the
38 * name replaces the job.
39 * @default a generated physical name
40 */
41 jobName?: string;
42 /**
43 * The IAM role (ARN or name) the job assumes to run.
44 */
45 role: string;
46 /**
47 * The job's execution command (script location + runtime).
48 */
49 command: JobCommand;
50 /**
51 * A description of the job.
52 */
53 description?: string;
54 /**
55 * Default arguments passed to the script (keys are `--`-prefixed), e.g.
56 * `{ "--TempDir": "s3://.../tmp/", "--job-language": "python" }`.
57 */
58 defaultArguments?: Record<string, string>;
59 /**
60 * Arguments that cannot be overridden at run time.
61 */
62 nonOverridableArguments?: Record<string, string>;
63 /**
64 * Glue connection names the job uses.
65 */
66 connections?: string[];
67 /**
68 * Max retries before the job run is considered failed.
69 */
70 maxRetries?: number;
71 /**
72 * Job timeout, e.g. `"1 hour"` or `Duration.minutes(60)`. Rounded to
73 * whole minutes on the wire (the Glue API unit).
74 */
75 timeout?: Duration.Input;
76 /**
77 * Max Glue data processing units (DPUs) for `pythonshell` (0.0625 or 1).
78 * Mutually exclusive with `numberOfWorkers`/`workerType`.
79 */
80 maxCapacity?: number;
81 /**
82 * The Glue version (e.g. `4.0`, `3.0`).
83 */
84 glueVersion?: string;
85 /**
86 * The number of workers of `workerType` allocated (Spark jobs).
87 */
88 numberOfWorkers?: number;
89 /**
90 * The worker type for Spark jobs: `Standard`, `G.1X`, `G.2X`, `G.025X`,
91 * `G.4X`, `G.8X`, `Z.2X`.
92 */
93 workerType?: string;
94 /**
95 * Execution class: `STANDARD` or `FLEX`.
96 */
97 executionClass?: "STANDARD" | "FLEX";
98 /**
99 * Concurrency configuration.
100 */
101 executionProperty?: {
102 /** Maximum number of concurrent runs allowed. */
103 maxConcurrentRuns?: number;
104 };
105 /**
106 * Tags to apply to the job. Merged with internal Alchemy tags.
107 */
108 tags?: Record<string, string>;
109}
110
111export interface Job extends Resource<
112 "AWS.Glue.Job",
113 JobProps,
114 {
115 /** The name of the job. */
116 jobName: string;
117 /** The ARN of the job. */
118 jobArn: string;
119 /** The IAM role the job assumes to run. */
120 role: string;
121 },
122 {},
123 Providers
124> {}
125
126/**
127 * An AWS Glue job — a Spark (`glueetl`), Python shell (`pythonshell`), or
128 * streaming ETL job definition (script in S3 + IAM role + arguments). The
129 * definition lifecycle is instant and free; job *runs* are billed and are
130 * started via `startJobRun`.
131 * ### Creating Jobs
132 * **Example:** Python Shell Job
133 * ```typescript
134 * import * as AWS from "alchemy/AWS";
135 *
136 * const job = yield* AWS.Glue.Job("Etl", {
137 * role: jobRole.roleArn,
138 * command: {
139 * name: "pythonshell",
140 * pythonVersion: "3.9",
141 * scriptLocation: "s3://my-bucket/scripts/etl.py",
142 * },
143 * maxCapacity: 0.0625,
144 * glueVersion: "3.0",
145 * defaultArguments: { "--job-language": "python" },
146 * });
147 * ```
148 *
149 * **Example:** Spark ETL Job
150 * ```typescript
151 * const job = yield* AWS.Glue.Job("SparkEtl", {
152 * role: jobRole.roleArn,
153 * command: {
154 * name: "glueetl",
155 * scriptLocation: "s3://my-bucket/scripts/spark.py",
156 * },
157 * glueVersion: "4.0",
158 * workerType: "G.1X",
159 * numberOfWorkers: 2,
160 * timeout: "1 hour",
161 * });
162 * ```
163 *
164 * ### Running Jobs
165 * **Example:** Start a Job Run from a Lambda
166 * ```typescript
167 * // init
168 * const startJobRun = yield* AWS.Glue.StartJobRun(job);
169 *
170 * // runtime
171 * const { JobRunId } = yield* startJobRun({});
172 * ```
173 *
174 * @resource
175 */
176export const Job = Resource<Job>("AWS.Glue.Job");
177
178export const JobProvider = () =>
179 Provider.effect(
180 Job,
181 Effect.gen(function* () {
182 const createName = Effect.fn(function* (
183 id: string,
184 props: { jobName?: string | undefined },
185 ) {
186 return (
187 props.jobName ?? (yield* createPhysicalName({ id, maxLength: 255 }))
188 );
189 });
190
191 const observe = Effect.fn(function* (name: string) {
192 return yield* glue.getJob({ JobName: name }).pipe(
193 Effect.map((r) => r.Job),
194 Effect.catchTag("EntityNotFoundException", () =>
195 Effect.succeed(undefined),
196 ),
197 );
198 });
199
200 const buildCommand = (command: JobCommand) => ({
201 Name: command.name,
202 ScriptLocation: command.scriptLocation,
203 PythonVersion: command.pythonVersion,
204 Runtime: command.runtime,
205 });
206
207 const buildDefinition = (props: JobProps) => ({
208 Role: props.role,
209 Command: buildCommand(props.command),
210 Description: props.description,
211 DefaultArguments: props.defaultArguments,
212 NonOverridableArguments: props.nonOverridableArguments,
213 Connections:
214 props.connections !== undefined
215 ? { Connections: props.connections }
216 : undefined,
217 MaxRetries: props.maxRetries,
218 Timeout: toWireMinutes(props.timeout),
219 MaxCapacity: props.maxCapacity,
220 GlueVersion: props.glueVersion,
221 NumberOfWorkers: props.numberOfWorkers,
222 WorkerType: props.workerType,
223 ExecutionClass: props.executionClass,
224 ExecutionProperty: props.executionProperty
225 ? { MaxConcurrentRuns: props.executionProperty.maxConcurrentRuns }
226 : undefined,
227 });
228
229 return Job.Provider.of({
230 stables: ["jobName", "jobArn"],
231
232 list: () =>
233 Effect.gen(function* () {
234 const { accountId, region } = yield* AWSEnvironment.current;
235 const pages = yield* glue.getJobs.pages({}).pipe(Stream.runCollect);
236 return Array.from(pages)
237 .flatMap((page) => page.Jobs ?? [])
238 .filter((j) => j.Name !== undefined)
239 .map((j) => ({
240 jobName: j.Name!,
241 jobArn: jobArn(region, accountId, j.Name!),
242 role: j.Role ?? "",
243 }));
244 }),
245
246 read: Effect.fn(function* ({ id, olds, output }) {
247 const { accountId, region } = yield* AWSEnvironment.current;
248 const name = output?.jobName ?? (yield* createName(id, olds ?? {}));
249 const job = yield* observe(name);
250 if (job?.Name === undefined) return undefined;
251 const arn = jobArn(region, accountId, job.Name);
252 const attrs = {
253 jobName: job.Name,
254 jobArn: arn,
255 role: job.Role ?? "",
256 };
257 const tags = yield* fetchObservedTags(arn);
258 return (yield* hasAlchemyTags(id, tags)) ? attrs : Unowned(attrs);
259 }),
260
261 diff: Effect.fn(function* ({ id, news, olds }) {
262 if (!isResolved(news)) return undefined;
263 const oldName = yield* createName(id, olds);
264 const newName = yield* createName(id, news);
265 if (oldName !== newName) return { action: "replace" } as const;
266 // everything else is UpdateJob-able
267 }),
268
269 reconcile: Effect.fn(function* ({ id, news, output, session }) {
270 const { accountId, region } = yield* AWSEnvironment.current;
271 const name = output?.jobName ?? (yield* createName(id, news));
272 const internalTags = yield* createInternalTags(id);
273 const desiredTags = { ...news.tags, ...internalTags };
274 const arn = jobArn(region, accountId, name);
275
276 // 1. OBSERVE
277 let job = yield* observe(name);
278
279 // 2. ENSURE / 3. SYNC
280 if (job === undefined) {
281 yield* retryWhileRoleNotAssumable(
282 glue.createJob({
283 Name: name,
284 ...buildDefinition(news),
285 Tags: desiredTags,
286 }),
287 ).pipe(
288 Effect.catchTag("AlreadyExistsException", () => Effect.void),
289 );
290 } else {
291 // UpdateJob replaces the full JobUpdate (Name is not part of it).
292 yield* retryWhileRoleNotAssumable(
293 glue.updateJob({
294 JobName: name,
295 JobUpdate: buildDefinition(news),
296 }),
297 );
298 }
299
300 // 3b. SYNC TAGS
301 const observedTags = yield* fetchObservedTags(arn);
302 yield* syncTags(arn, observedTags, desiredTags);
303
304 job = yield* observe(name);
305
306 yield* session.note(name);
307 return {
308 jobName: name,
309 jobArn: arn,
310 role: job?.Role ?? news.role,
311 };
312 }),
313
314 delete: Effect.fn(function* ({ output }) {
315 // DeleteJob is idempotent — the API returns success (not an error)
316 // when the job definition is already gone.
317 yield* glue.deleteJob({ JobName: output.jobName });
318 }),
319 });
320 }),
321 );