Skip to main content
Version: 5.17

Tutorial: Run a data pipeline using Jet

Learn how to run a data pipeline in Hazelcast using Jet.

Overview​

In this tutorial, you'll:

  • Build and deploy a custom Jet pipeline implementation to cloud storage.
  • Deploy Hazelcast with the Jet engine configured.
  • Start the Jet pipeline.
  • Check the Jet pipeline.
  • Cancel the Jet pipeline.

The tutorial should take approximately 20 minutes to complete.

Prerequisites​

Before you begin, make sure that you have:

Deploy the Jet pipeline JAR to cloud storage​

In this step, you'll build the Jet pipeline JAR from the tutorial's GitHub repository, and upload it to your cloud provider.

  1. Clone the sample project.

    bash
    git clone https://github.com/hazelcast-guides/hazelcast-platform-operator-jet.git

    cd hazelcast-platform-operator-jet

    The sample code for this tutorial is in the jet-pipeline/src/main/java/org/example/run/ directory.

  2. Build the Jet pipeline JAR.

    shell
    mvn package -f ./jet-pipeline

  3. Upload the Jet pipeline JAR to the storage blob/bucket of your cloud provider, replacing the placeholder values.

    bash
    aws s3 cp jet-pipeline/target/jet-pipeline-run-1.0.0.jar s3://<BUCKET_NAME>

  4. Create a secret for your cloud storage.

    bash
    kubectl create secret generic <SECRET-NAME> --from-literal=region=<region> \
    	--from-literal=access-key-id=<access-key-id> \
    	--from-literal=secret-access-key=<secret-access-key>
tip

If you want to learn more about other authentication methods, see Authorization Methods to Access Cloud Storage.

Start the Hazelcast cluster​

  1. Create a secret with your Hazelcast Enterprise License.

    shell
    kubectl create secret generic hazelcast-license-key --from-literal=license-key=<hz-license-key>

  2. Create a Hazelcast cluster with the Jet engine configured.

    shell
    kubectl apply -f - <<EOF
    apiVersion: hazelcast.com/v1alpha1
    kind: Hazelcast
    metadata:
      name: my-hazelcast
    spec:
      licenseKeySecretName: hazelcast-license-key
      clusterSize: 3
      jet:
        resourceUploadEnabled: true
    EOF
    note

    You need to use Hazelcast 5.3 or above to submit Jet jobs with Operator.

  3. Check the cluster status to make sure that it's up and running.

    shell
    kubectl get hazelcast my-hazelcast
    NAME           STATUS    MEMBERS
    my-hazelcast   Running   3/3

Run the data pipeline​

In this step, you'll start, check and cancel the Jet job that runs the data pipeline.

  1. Start the Jet job you have already deployed:

    Replace the <BUCKET-URI> with the appropriate sample bucket URI for your cloud provider:

    • s3://hazelcast-jet-pipeline
    • gs://hazelcast-jet-pipeline
    • azblob://hazelcast-jet-pipeline
    shell
    kubectl apply -f - <<EOF
    apiVersion: hazelcast.com/v1alpha1
    kind: JetJob
    metadata:
      name: my-jet-job
    spec:
      hazelcastResourceName: my-hazelcast
      state: Running
      jarName: jet-pipeline-run-1.0.0.jar
      bucketConfig:
        bucketURI: "<BUCKET-URI>"
        secretName: <SECRET-NAME>
    EOF

  2. Check status of the Jet job.

    bash
    kubectl get jetjob my-jet-job -w

    The output will look something like this. Wait until the Running status appears.

    bash
    NAME         STATUS    ID                   SUBMISSIONTIME         COMPLETIONTIME
    my-jet-job   Starting  0
    my-jet-job   Running   732187341048774657   2023-07-14T10:51:06Z

  3. Check the output of the Jet pipeline in the logs:

    bash
    kubectl logs -l app.kubernetes.io/name=hazelcast -c hazelcast | grep sequence

    The logs of the Jet pipeline are displayed in the output.

    bash
    {"time":"2023-07-14T10:51:06,679", "logger": "com.hazelcast.jet.impl.connector.WriteLoggerP", "level": "INFO", "msg": "[172.17.0.6]:5702 [dev] [5.3.1] [my-jet-job/loggerSink#0] SimpleEvent(timestamp=10:51:06.000, sequence=0) "}
    {"time":"2023-07-14T10:51:07,675", "logger": "com.hazelcast.jet.impl.connector.WriteLoggerP", "level": "INFO", "msg": "[172.17.0.6]:5702 [dev] [5.3.1] [my-jet-job/loggerSink#0] SimpleEvent(timestamp=10:51:07.000, sequence=1) "}
    {"time":"2023-07-14T10:51:08,681", "logger": "com.hazelcast.jet.impl.connector.WriteLoggerP", "level": "INFO", "msg": "[172.17.0.6]:5702 [dev] [5.3.1] [my-jet-job/loggerSink#0] SimpleEvent(timestamp=10:51:08.000, sequence=2) "}
    {"time":"2023-07-14T10:51:09,679", "logger": "com.hazelcast.jet.impl.connector.WriteLoggerP", "level": "INFO", "msg": "[172.17.0.6]:5702 [dev] [5.3.1] [my-jet-job/loggerSink#0] SimpleEvent(timestamp=10:51:09.000, sequence=3) "}
    {"time":"2023-07-14T10:51:10,682", "logger": "com.hazelcast.jet.impl.connector.WriteLoggerP", "level": "INFO", "msg": "[172.17.0.6]:5702 [dev] [5.3.1] [my-jet-job/loggerSink#0] SimpleEvent(timestamp=10:51:10.000, sequence=4) "}
    {"time":"2023-07-14T10:51:11,678", "logger": "com.hazelcast.jet.impl.connector.WriteLoggerP", "level": "INFO", "msg": "[172.17.0.6]:5702 [dev] [5.3.1] [my-jet-job/loggerSink#0] SimpleEvent(timestamp=10:51:11.000, sequence=5) "}
    {"time":"2023-07-14T10:51:12,677", "logger": "com.hazelcast.jet.impl.connector.WriteLoggerP", "level": "INFO", "msg": "[172.17.0.6]:5702 [dev] [5.3.1] [my-jet-job/loggerSink#0] SimpleEvent(timestamp=10:51:12.000, sequence=6) "}
    ....
    ....

  4. Cancel the Jet job by setting the spec.state to Canceled. Use one of the following options.

    1. Cancel the Jet job using kubectl apply:

      shell
      kubectl apply -f - <<EOF
      apiVersion: hazelcast.com/v1alpha1
      kind: JetJob
      metadata:
        name: my-jet-job
      spec:
        hazelcastResourceName: my-hazelcast
        state: Canceled
        jarName: jet-pipeline-run-1.0.0.jar
        bucketConfig:
          bucketURI: "<BUCKET-URI>"
          secretName: <SECRET-NAME>
      EOF

    2. Cancel the Jet job using kubectl patch:

      bash
      kubectl patch jetjob my-jet-job -p '{"spec":{"state":"Canceled"}}' --type=merge

  5. Check the Jet job status to make sure the job was cancelled.

    bash
    kubectl get jetjob my-jet-job

    The output shows the Jet job status as ExecutionFailed.

    bash
    NAME         STATUS            ID                   SUBMISSIONTIME         COMPLETIONTIME
    my-jet-job   ExecutionFailed   732229926463209474   2023-07-14T13:40:19Z   2023-07-14T14:07:51Z

  6. See a more detailed message for the status of the Jet job:

    bash
    kubectl get jetjob my-jet-job -o custom-columns=:.status.failureText

    The failure text appears in the output.

    bash
    com.hazelcast.jet.impl.exception.CancellationByUserException
          at com.hazelcast.jet.impl.MasterJobContext.createCancellationException(MasterJobContext.java:211)
          at com.hazelcast.jet.impl.MasterJobContext.getErrorFromResponses(MasterJobContext.java:653)
          at com.hazelcast.jet.impl.MasterJobContext.lambda$invokeStartExecution$12(MasterJobContext.java:576)
          at com.hazelcast.jet.impl.MasterContext.lambda$invokeOnParticipant$3(MasterContext.java:376)
          ....
          ....

Clean up​

To clean up all the resources you created during the tutorial, and to remove the custom resources and secrets, run the following command:

shell
kubectl delete secret <SECRET-NAME>
kubectl delete $(kubectl get hazelcast -o name)
note

Deleting the Hazelcast CR also deletes the Jet jobs linked to it.

Summary​

In this tutorial, you learned how to:

  • Build and deploy a custom Jet pipeline implementation to cloud storage.
  • Deploy Hazelcast with the Jet engine configured.
  • Start a Jet pipeline.
  • Check a Jet pipeline.
  • Cancel a Jet pipeline.

Next steps​

See also:

Edit this page