• Deploy code to python shell / pyspark AWS Glue jobs.
• Use AWS Sagemaker to create ML Models.
• Orchestrate the above jobs using AWS Stepfunctions as simple as task1 >> task2

• Let us know what you want to see next.

  
  

Installation

Datajob can be installed using pip.
Beware that we depend on aws cdk cli!

pip install datajob
npm install -g aws-cdk@1.109.0 # latest version of datajob depends this version

Quickstart

You can find the full example in examples/data_pipeline_simple.

We have a simple data pipeline composed of 2 glue jobs orchestrated sequentially using step functions.

from aws_cdk import core

from datajob.datajob_stack import DataJobStack
from datajob.glue.glue_job import GlueJob
from datajob.stepfunctions.stepfunctions_workflow import StepfunctionsWorkflow

app = core.App()

# The datajob_stack is the instance that will result in a cloudformation stack.
# We inject the datajob_stack object through all the resources that we want to add.
with DataJobStack(scope=app, id="data-pipeline-simple") as datajob_stack:
    # We define 2 glue jobs with the relative path to the source code.
    task1 = GlueJob(
        datajob_stack=datajob_stack, name="task1", job_path="glue_jobs/task.py"
    )
    task2 = GlueJob(
        datajob_stack=datajob_stack, name="task2", job_path="glue_jobs/task2.py"
    )

    # We instantiate a step functions workflow and orchestrate the glue jobs.
    with StepfunctionsWorkflow(datajob_stack=datajob_stack, name="workflow") as sfn:
        task1 >> task2

app.synth()

We add the above code in a file called datajob_stack.py in the root of the project.

Configure CDK

Follow the steps here to configure your credentials.

export AWS_PROFILE=default
# use the aws cli to get your account number
export AWS_ACCOUNT=$(aws sts get-caller-identity --query Account --output text --profile $AWS_PROFILE)
export AWS_DEFAULT_REGION=eu-west-1

# init cdk
cdk bootstrap aws://$AWS_ACCOUNT/$AWS_DEFAULT_REGION

Deploy

Deploy the pipeline using CDK.

cd examples/data_pipeline_simple
cdk deploy --app  "python datajob_stack.py" --require-approval never

Execute

datajob execute --state-machine data-pipeline-simple-workflow

The terminal will show a link to the step functions page to follow up on your pipeline run.

Destroy

cdk destroy --app  "python datajob_stack.py"

Examples

• Data pipeline with parallel steps
• Data pipeline for processing big data using PySpark
• Data pipeline where you package and ship your project as a wheel
• Machine Learning pipeline where we combine glue jobs with sagemaker

All our examples are in ./examples

Functionality

Datajob in depth

The datajob_stack is the instance that will result in a cloudformation stack. The path in project_root helps datajob_stack locate the root of the project where the setup.py/poetry pyproject.toml file can be found, as well as the dist/ folder with the wheel of your project .

import pathlib
from aws_cdk import core

from datajob.datajob_stack import DataJobStack

current_dir = pathlib.Path(__file__).parent.absolute()
app = core.App()

with DataJobStack(
    scope=app, id="data-pipeline-pkg", project_root=current_dir
) as datajob_stack:

    ...

When entering the contextmanager of DataJobStack:

A DataJobContext is initialized to deploy and run a data pipeline on AWS. The following resources are created: 1) "data bucket"

• an S3 bucket that you can use to dump ingested data, dump intermediate results and the final output.
• you can access the data bucket as a Bucket object via datajob_stack.context.data_bucket
• you can access the data bucket name via datajob_stack.context.data_bucket_name 2) "deployment bucket"
  • an s3 bucket to deploy code, artifacts, scripts, config, files, ...
  • you can access the deployment bucket as a Bucket object via datajob_stack.context.deployment_bucket
  • you can access the deployment bucket name via datajob_stack.context.deployment_bucket_name

when exiting the context manager all the resources of our DataJobStack object are created.

Ideas

Any suggestions can be shared by starting a discussion

These are the ideas, we find interesting to implement;

• add a time based trigger to the step functions workflow.
• add an s3 event trigger to the step functions workflow.
• add a lambda that copies data from one s3 location to another.
• version your data pipeline.
• cli command to view the logs / glue jobs / s3 bucket
• implement sagemaker services
  • processing jobs
  • hyperparameter tuning jobs
  • training jobs
• implement lambda
• implement ECS Fargate
• create a serverless UI that follows up on the different pipelines deployed on possibly different AWS accounts using Datajob

Feedback is much appreciated!