1import * as databrew from "@distilled.cloud/aws/databrew";
2import * as Data from "effect/Data";
3import type * as Duration from "effect/Duration";
4import * as Effect from "effect/Effect";
5import * as Option from "effect/Option";
6import * as Schedule from "effect/Schedule";
7import * as Stream from "effect/Stream";
8import { Unowned } from "../../AdoptPolicy.ts";
9import { isResolved } from "../../Diff.ts";
10import { createPhysicalName } from "../../PhysicalName.ts";
11import * as Provider from "../../Provider.ts";
12import { Resource } from "../../Resource.ts";
13import { createInternalTags, hasAlchemyTags } from "../../Tags.ts";
14import { toWireMinutes } from "../../Util/Duration.ts";
15import { AWSEnvironment } from "../Environment.ts";
16import type { Providers } from "../Providers.ts";
17import { buildS3Location, type S3Location } from "./Dataset.ts";
18import {
19 cleanMap,
20 databrewArn,
21 fetchObservedTags,
22 retryWhileConflict,
23 retryWhileRoleNotAssumable,
24 syncTags,
25} from "./internal.ts";
26
27/** An S3 output target for a recipe job. */
28export interface JobOutput {
29 /** The S3 location the transformed data is written to. */
30 location: S3Location;
31 /** The output file format. @default "CSV" */
32 format?:
33 | "CSV"
34 | "JSON"
35 | "PARQUET"
36 | "GLUEPARQUET"
37 | "AVRO"
38 | "ORC"
39 | "XML"
40 | "TABLEAUHYPER"
41 | (string & {});
42 /** Compression applied to the output files. */
43 compressionFormat?:
44 | "GZIP"
45 | "LZ4"
46 | "SNAPPY"
47 | "BZIP2"
48 | "DEFLATE"
49 | "LZO"
50 | "BROTLI"
51 | "ZSTD"
52 | "ZLIB"
53 | (string & {});
54 /** Columns to partition the output by. */
55 partitionColumns?: string[];
56 /** Overwrite previous output files on each run. @default false */
57 overwrite?: boolean;
58 /** Format-specific output options. */
59 formatOptions?: {
60 /** CSV output options. */
61 csv?: {
62 /** The single-character field delimiter. @default "," */
63 delimiter?: string;
64 };
65 };
66 /** Maximum number of output files. */
67 maxOutputFiles?: number;
68}
69
70/** A pointer to the recipe (and version) a recipe job applies. */
71export interface JobRecipeReference {
72 /** The recipe name. */
73 name: string;
74 /**
75 * The recipe version to run.
76 * @default "LATEST_PUBLISHED"
77 */
78 recipeVersion?: string;
79}
80
81/** How much of the dataset a profile job analyzes. */
82export interface JobSample {
83 /** `FULL_DATASET` or `CUSTOM_ROWS`. @default "CUSTOM_ROWS" */
84 mode?: "FULL_DATASET" | "CUSTOM_ROWS" | (string & {});
85 /** Row count when mode is `CUSTOM_ROWS`. @default 20000 */
86 size?: number;
87}
88
89/** Selects columns by exact name or regex. */
90export interface ColumnSelector {
91 /** A regular expression matching column names. */
92 regex?: string;
93 /** An exact column name. */
94 name?: string;
95}
96
97/** Which statistics a profile job computes. */
98export interface StatisticsConfiguration {
99 /** Statistics to include (default: all supported). */
100 includedStatistics?: string[];
101 /** Per-statistic parameter overrides. */
102 overrides?: {
103 /** The statistic name. */
104 statistic: string;
105 /** Statistic parameters. */
106 parameters: Record<string, string>;
107 }[];
108}
109
110/** Fine-grained configuration for a profile job. */
111export interface ProfileJobConfiguration {
112 /** Dataset-level statistics configuration. */
113 datasetStatisticsConfiguration?: StatisticsConfiguration;
114 /** Restrict profiling to these columns. */
115 profileColumns?: ColumnSelector[];
116 /** Column-level statistics configurations. */
117 columnStatisticsConfigurations?: {
118 /** The columns the configuration applies to (default: all). */
119 selectors?: ColumnSelector[];
120 /** The statistics to compute for those columns. */
121 statistics: StatisticsConfiguration;
122 }[];
123 /** PII entity detection configuration. */
124 entityDetectorConfiguration?: {
125 /** Entity types to detect, e.g. `["USA_SSN", "EMAIL"]`. */
126 entityTypes: string[];
127 /** Statistics allowed on detected-entity columns. */
128 allowedStatistics?: {
129 /** The allowed statistic names. */
130 statistics: string[];
131 }[];
132 };
133}
134
135/** Attaches a ruleset to a profile job for data-quality validation. */
136export interface ValidationConfiguration {
137 /** The ARN of the ruleset to validate against. */
138 rulesetArn: string;
139 /** The validation mode. @default "CHECK_ALL" */
140 validationMode?: "CHECK_ALL" | (string & {});
141}
142
143export interface JobProps {
144 /**
145 * Name of the job. If omitted, a unique name is generated. Changing the
146 * name replaces the job.
147 * @default a generated physical name
148 */
149 jobName?: string;
150 /**
151 * The job type: `PROFILE` analyzes a dataset and writes a data profile;
152 * `RECIPE` applies a recipe's transformations and writes the output.
153 * Changing the type replaces the job.
154 */
155 type: "PROFILE" | "RECIPE";
156 /**
157 * The dataset the job reads. Required for `PROFILE` jobs; for `RECIPE`
158 * jobs provide either `datasetName` + `recipeReference` or `projectName`.
159 * Changing it replaces the job.
160 */
161 datasetName?: string;
162 /**
163 * The IAM role ARN DataBrew assumes to run the job (read the input,
164 * write the output).
165 */
166 role: string;
167 /**
168 * S3 location the profile results are written to (PROFILE jobs only).
169 */
170 outputLocation?: S3Location;
171 /**
172 * Fine-grained profiling configuration (PROFILE jobs only).
173 */
174 configuration?: ProfileJobConfiguration;
175 /**
176 * Rulesets to validate the data against (PROFILE jobs only).
177 */
178 validationConfigurations?: ValidationConfiguration[];
179 /**
180 * How much of the dataset to profile (PROFILE jobs only).
181 * @default 20,000 rows
182 */
183 jobSample?: JobSample;
184 /**
185 * S3 outputs the transformed data is written to (RECIPE jobs only).
186 */
187 outputs?: JobOutput[];
188 /**
189 * The recipe to apply (RECIPE jobs only). Defaults to the recipe's latest
190 * *published* version. Changing it replaces the job.
191 */
192 recipeReference?: JobRecipeReference;
193 /**
194 * Derive dataset + recipe from an existing DataBrew project instead of
195 * `datasetName`/`recipeReference` (RECIPE jobs only). Changing it
196 * replaces the job.
197 */
198 projectName?: string;
199 /**
200 * The encryption mode for job output: `SSE-KMS` or `SSE-S3`.
201 */
202 encryptionMode?: "SSE-KMS" | "SSE-S3";
203 /**
204 * The KMS key ARN when `encryptionMode` is `SSE-KMS`.
205 */
206 encryptionKeyArn?: string;
207 /**
208 * Enable CloudWatch logging for the job.
209 * @default "ENABLE"
210 */
211 logSubscription?: "ENABLE" | "DISABLE";
212 /**
213 * Maximum number of compute nodes the job can consume.
214 * @default 5
215 */
216 maxCapacity?: number;
217 /**
218 * Maximum retries after a job run fails.
219 * @default 0
220 */
221 maxRetries?: number;
222 /**
223 * Job timeout, e.g. `"90 minutes"` or `Duration.hours(2)`. DataBrew
224 * measures the timeout in whole minutes.
225 * @default "48 hours"
226 */
227 timeout?: Duration.Input;
228 /**
229 * Tags to apply to the job. Merged with internal Alchemy tags.
230 */
231 tags?: Record<string, string>;
232}
233
234export interface Job extends Resource<
235 "AWS.DataBrew.Job",
236 JobProps,
237 {
238 /** Name of the job. */
239 jobName: string;
240 /** ARN of the job. */
241 jobArn: string;
242 /** Type of the job (`PROFILE` or `RECIPE`). */
243 type: string;
244 },
245 {},
246 Providers
247> {}
248
249/**
250 * An AWS Glue DataBrew job definition — either a `PROFILE` job that analyzes
251 * a dataset and writes a data-quality profile to S3, or a `RECIPE` job that
252 * applies a published recipe's transformations and writes the result to S3.
253 * The definition is free and instant; job *runs* are billed per node-hour
254 * and are started with `StartJobRun`.
255 * ### Profile Jobs
256 * **Example:** Profile a Dataset
257 * ```typescript
258 * import * as AWS from "alchemy/AWS";
259 *
260 * const profile = yield* AWS.DataBrew.Job("Profile", {
261 * type: "PROFILE",
262 * datasetName: dataset.datasetName,
263 * role: role.roleArn,
264 * outputLocation: { bucket: bucket.bucketName, key: "profiles/" },
265 * jobSample: { mode: "CUSTOM_ROWS", size: 1000 },
266 * });
267 * ```
268 *
269 * ### Recipe Jobs
270 * **Example:** Transform with a Published Recipe
271 * ```typescript
272 * const transform = yield* AWS.DataBrew.Job("Transform", {
273 * type: "RECIPE",
274 * datasetName: dataset.datasetName,
275 * recipeReference: { name: recipe.recipeName },
276 * role: role.roleArn,
277 * outputs: [
278 * {
279 * location: { bucket: bucket.bucketName, key: "curated/" },
280 * format: "CSV",
281 * overwrite: true,
282 * },
283 * ],
284 * });
285 * ```
286 *
287 * @resource
288 */
289export const Job = Resource<Job>("AWS.DataBrew.Job");
290
291/** The job's props don't satisfy the requirements of its `type`. */
292export class DataBrewJobConfigError extends Data.TaggedError(
293 "DataBrewJobConfigError",
294)<{ message: string }> {}
295
296const validate = (props: JobProps) => {
297 if (props.type === "PROFILE") {
298 if (props.datasetName === undefined || props.outputLocation === undefined) {
299 return Effect.fail(
300 new DataBrewJobConfigError({
301 message: "PROFILE jobs require `datasetName` and `outputLocation`.",
302 }),
303 );
304 }
305 if (props.outputs !== undefined || props.recipeReference !== undefined) {
306 return Effect.fail(
307 new DataBrewJobConfigError({
308 message: "`outputs`/`recipeReference` only apply to RECIPE jobs.",
309 }),
310 );
311 }
312 } else {
313 const viaProject = props.projectName !== undefined;
314 const viaRecipe =
315 props.datasetName !== undefined && props.recipeReference !== undefined;
316 if (!viaProject && !viaRecipe) {
317 return Effect.fail(
318 new DataBrewJobConfigError({
319 message:
320 "RECIPE jobs require either `projectName` or `datasetName` + `recipeReference`.",
321 }),
322 );
323 }
324 if (props.outputs === undefined || props.outputs.length === 0) {
325 return Effect.fail(
326 new DataBrewJobConfigError({
327 message: "RECIPE jobs require at least one entry in `outputs`.",
328 }),
329 );
330 }
331 if (
332 props.outputLocation !== undefined ||
333 props.configuration !== undefined ||
334 props.jobSample !== undefined ||
335 props.validationConfigurations !== undefined
336 ) {
337 return Effect.fail(
338 new DataBrewJobConfigError({
339 message:
340 "`outputLocation`/`configuration`/`jobSample`/`validationConfigurations` only apply to PROFILE jobs.",
341 }),
342 );
343 }
344 }
345 return Effect.void;
346};
347
348const buildOutputs = (outputs: JobOutput[] | undefined) =>
349 outputs?.map((output) => ({
350 Location: buildS3Location(output.location),
351 Format: output.format,
352 CompressionFormat: output.compressionFormat,
353 PartitionColumns: output.partitionColumns,
354 Overwrite: output.overwrite,
355 FormatOptions: output.formatOptions
356 ? {
357 Csv: output.formatOptions.csv
358 ? { Delimiter: output.formatOptions.csv.delimiter }
359 : undefined,
360 }
361 : undefined,
362 MaxOutputFiles: output.maxOutputFiles,
363 }));
364
365const buildStatisticsConfiguration = (config: StatisticsConfiguration) => ({
366 IncludedStatistics: config.includedStatistics,
367 Overrides: config.overrides?.map((o) => ({
368 Statistic: o.statistic,
369 Parameters: o.parameters,
370 })),
371});
372
373const buildColumnSelectors = (selectors: ColumnSelector[] | undefined) =>
374 selectors?.map((s) => ({ Regex: s.regex, Name: s.name }));
375
376const buildConfiguration = (config: ProfileJobConfiguration | undefined) =>
377 config
378 ? {
379 DatasetStatisticsConfiguration: config.datasetStatisticsConfiguration
380 ? buildStatisticsConfiguration(config.datasetStatisticsConfiguration)
381 : undefined,
382 ProfileColumns: buildColumnSelectors(config.profileColumns),
383 ColumnStatisticsConfigurations:
384 config.columnStatisticsConfigurations?.map((c) => ({
385 Selectors: buildColumnSelectors(c.selectors),
386 Statistics: buildStatisticsConfiguration(c.statistics),
387 })),
388 EntityDetectorConfiguration: config.entityDetectorConfiguration
389 ? {
390 EntityTypes: config.entityDetectorConfiguration.entityTypes,
391 AllowedStatistics:
392 config.entityDetectorConfiguration.allowedStatistics?.map(
393 (a) => ({ Statistics: a.statistics }),
394 ),
395 }
396 : undefined,
397 }
398 : undefined;
399
400const buildValidationConfigurations = (
401 configs: ValidationConfiguration[] | undefined,
402) =>
403 configs?.map((c) => ({
404 RulesetArn: c.rulesetArn,
405 ValidationMode: c.validationMode,
406 }));
407
408const buildJobSample = (sample: JobSample | undefined) =>
409 sample ? { Mode: sample.mode, Size: sample.size } : undefined;
410
411const buildCommon = (props: JobProps) => ({
412 EncryptionKeyArn: props.encryptionKeyArn,
413 EncryptionMode: props.encryptionMode,
414 LogSubscription: props.logSubscription,
415 MaxCapacity: props.maxCapacity,
416 MaxRetries: props.maxRetries,
417 RoleArn: props.role,
418 Timeout: toWireMinutes(props.timeout),
419});
420
421export const JobProvider = () =>
422 Provider.effect(
423 Job,
424 Effect.gen(function* () {
425 const createName = Effect.fn(function* (
426 id: string,
427 props: { jobName?: string | undefined },
428 ) {
429 return (
430 props.jobName ?? (yield* createPhysicalName({ id, maxLength: 240 }))
431 );
432 });
433
434 const observe = Effect.fn(function* (name: string) {
435 return yield* databrew
436 .describeJob({ Name: name })
437 .pipe(
438 Effect.catchTag("ResourceNotFoundException", () =>
439 Effect.succeed(undefined),
440 ),
441 );
442 });
443
444 return Job.Provider.of({
445 stables: ["jobName", "jobArn"],
446
447 list: () =>
448 Effect.gen(function* () {
449 const { accountId, region } = yield* AWSEnvironment.current;
450 const pages = yield* databrew.listJobs
451 .pages({})
452 .pipe(Stream.runCollect);
453 return Array.from(pages)
454 .flatMap((page) => page.Jobs ?? [])
455 .map((j) => ({
456 jobName: j.Name,
457 jobArn:
458 j.ResourceArn ??
459 databrewArn(region, accountId, "job", j.Name),
460 type: j.Type ?? "RECIPE",
461 }));
462 }),
463
464 read: Effect.fn(function* ({ id, olds, output }) {
465 const { accountId, region } = yield* AWSEnvironment.current;
466 const name = output?.jobName ?? (yield* createName(id, olds ?? {}));
467 const job = yield* observe(name);
468 if (job === undefined) return undefined;
469 const arn =
470 job.ResourceArn ?? databrewArn(region, accountId, "job", name);
471 const attrs = {
472 jobName: name,
473 jobArn: arn,
474 type: job.Type ?? "RECIPE",
475 };
476 const tags = cleanMap(job.Tags);
477 return (yield* hasAlchemyTags(id, tags)) ? attrs : Unowned(attrs);
478 }),
479
480 diff: Effect.fn(function* ({ id, news, olds }) {
481 if (!isResolved(news)) return undefined;
482 const oldName = yield* createName(id, olds);
483 const newName = yield* createName(id, news);
484 if (oldName !== newName) return { action: "replace" } as const;
485 // The Update APIs accept neither the job type nor the input
486 // associations — those are create-only.
487 if (olds.type !== news.type) return { action: "replace" } as const;
488 if (olds.datasetName !== news.datasetName) {
489 return { action: "replace" } as const;
490 }
491 if (olds.projectName !== news.projectName) {
492 return { action: "replace" } as const;
493 }
494 if (
495 olds.recipeReference?.name !== news.recipeReference?.name ||
496 (olds.recipeReference?.recipeVersion ?? "LATEST_PUBLISHED") !==
497 (news.recipeReference?.recipeVersion ?? "LATEST_PUBLISHED")
498 ) {
499 return { action: "replace" } as const;
500 }
501 }),
502
503 reconcile: Effect.fn(function* ({ id, news, output, session }) {
504 const { accountId, region } = yield* AWSEnvironment.current;
505 yield* validate(news);
506 const name = output?.jobName ?? (yield* createName(id, news));
507 const internalTags = yield* createInternalTags(id);
508 const desiredTags = { ...news.tags, ...internalTags };
509
510 // 1. OBSERVE
511 const job = yield* observe(name);
512
513 // 2. ENSURE / 3. SYNC — Update{Profile,Recipe}Job is a full PUT
514 if (job === undefined) {
515 if (news.type === "PROFILE") {
516 yield* retryWhileRoleNotAssumable(
517 databrew.createProfileJob({
518 Name: name,
519 DatasetName: news.datasetName!,
520 OutputLocation: buildS3Location(news.outputLocation!),
521 Configuration: buildConfiguration(news.configuration),
522 ValidationConfigurations: buildValidationConfigurations(
523 news.validationConfigurations,
524 ),
525 JobSample: buildJobSample(news.jobSample),
526 Tags: desiredTags,
527 ...buildCommon(news),
528 }),
529 ).pipe(Effect.catchTag("ConflictException", () => Effect.void));
530 } else {
531 yield* retryWhileRoleNotAssumable(
532 databrew.createRecipeJob({
533 Name: name,
534 DatasetName: news.datasetName,
535 ProjectName: news.projectName,
536 RecipeReference: news.recipeReference
537 ? {
538 Name: news.recipeReference.name,
539 // the API rejects a null version ("version null is
540 // invalid") — apply the documented default explicitly
541 RecipeVersion:
542 news.recipeReference.recipeVersion ??
543 "LATEST_PUBLISHED",
544 }
545 : undefined,
546 Outputs: buildOutputs(news.outputs),
547 Tags: desiredTags,
548 ...buildCommon(news),
549 }),
550 ).pipe(Effect.catchTag("ConflictException", () => Effect.void));
551 }
552 } else if (news.type === "PROFILE") {
553 yield* retryWhileRoleNotAssumable(
554 databrew.updateProfileJob({
555 Name: name,
556 OutputLocation: buildS3Location(news.outputLocation!),
557 Configuration: buildConfiguration(news.configuration),
558 ValidationConfigurations: buildValidationConfigurations(
559 news.validationConfigurations,
560 ),
561 JobSample: buildJobSample(news.jobSample),
562 ...buildCommon(news),
563 }),
564 );
565 } else {
566 yield* retryWhileRoleNotAssumable(
567 databrew.updateRecipeJob({
568 Name: name,
569 Outputs: buildOutputs(news.outputs),
570 ...buildCommon(news),
571 }),
572 );
573 }
574
575 const arn =
576 job?.ResourceArn ?? databrewArn(region, accountId, "job", name);
577
578 // 3b. SYNC TAGS against observed cloud tags
579 const observedTags = yield* fetchObservedTags(arn);
580 yield* syncTags(arn, observedTags, desiredTags);
581
582 yield* session.note(name);
583 return { jobName: name, jobArn: arn, type: news.type };
584 }),
585
586 delete: Effect.fn(function* ({ output }) {
587 // A run in STARTING/RUNNING/STOPPING keeps the job's dataset and
588 // recipe associated — downstream Dataset/Recipe deletes then fail
589 // with ConflictException ("is used in job …") long after DeleteJob
590 // itself succeeds. Stop in-flight runs and wait (bounded) for every
591 // run to reach a terminal state before deleting the job.
592 const runs = yield* databrew.listJobRuns
593 .items({ Name: output.jobName })
594 .pipe(
595 Stream.runCollect,
596 Effect.map((chunk) => Array.from(chunk)),
597 Effect.catchTag("ResourceNotFoundException", () =>
598 Effect.succeed([]),
599 ),
600 );
601 const isActive = (state: string | undefined): boolean =>
602 state === "STARTING" || state === "RUNNING" || state === "STOPPING";
603 yield* Effect.forEach(
604 runs.filter(
605 (run) =>
606 run.RunId !== undefined &&
607 (run.State === "STARTING" || run.State === "RUNNING"),
608 ),
609 (run) =>
610 databrew
611 .stopJobRun({ Name: output.jobName, RunId: run.RunId! })
612 .pipe(
613 // Already stopping/stopped (ValidationException) or gone.
614 Effect.catchTag(
615 ["ResourceNotFoundException", "ValidationException"],
616 () => Effect.succeed(undefined),
617 ),
618 ),
619 );
620 if (runs.some((run) => isActive(run.State))) {
621 yield* databrew.listJobRuns.items({ Name: output.jobName }).pipe(
622 // Settled = no active run observed; stop paginating at the
623 // first active run instead of draining every page.
624 Stream.filter((run) => isActive(run.State)),
625 Stream.runHead,
626 Effect.map(Option.isNone),
627 Effect.catchTag("ResourceNotFoundException", () =>
628 Effect.succeed(true),
629 ),
630 Effect.repeat({
631 schedule: Schedule.spaced("5 seconds"),
632 until: (settled): boolean => settled,
633 times: 18,
634 }),
635 );
636 }
637 // ConflictException while a job run is still in flight.
638 yield* retryWhileConflict(
639 databrew.deleteJob({ Name: output.jobName }),
640 ).pipe(
641 Effect.catchTag("ResourceNotFoundException", () =>
642 Effect.succeed(undefined),
643 ),
644 );
645 }),
646 });
647 }),
648 );