Skip to main content

ConveyorSparkSubmitOperatorV2

The ConveyorSparkSubmitOperatorV2 provides an easy way to run a Spark job containing your transformations. The following is an example usage of the ConveyorSparkSubmitOperatorV2:

from conveyor.operators import ConveyorSparkSubmitOperatorV2

role = "job-role-{{ macros.conveyor.env() }}"

ConveyorSparkSubmitOperatorV2(
    task_id="the-task-id",
    num_executors=1,
    driver_instance_type="mx.small",
    executor_instance_type="mx.small",
    aws_role=role,
    java_class="name-of-the-class",
    application="path-to-jar-in-container",
    application_args=[
        "--environment", "{{ macros.conveyor.env() }}",
        "--snapshotDates", "2019-11-06"
    ]
)
info

The ConveyorSparkSubmitOperatorV2 extends the Airflow BaseOperator, so any parameter supported by the BaseOperator can also be set on the Conveyor operator. More detail on the supported parameters can be found here

The Conveyor specific parameters supported by the ConveyorSparkSubmitOperatorV2 are:

ParameterTypeDefaultExplanation
imagestrProject imageDocker image you wish to launch, the default is the image and version of your project. If you want to use the image of another project you can use macros.conveyor.image('PROJECT_NAME') (templated).
applicationstrThe application that submitted as a job, either a .jar or .py file (templated).
application_argslist[]Arguments for the application being submitted (templated).
confdictArbitrary Spark configuration properties (templated), see default settings below.
java_classstrThe main class of the Java application, this is not needed when using PySpark.
num_executorsint2Number of executors to launch (templated).
driver_instance_typestrmx.smallThe Conveyor instance type to use for the Spark driver. This specifies the CPU/Memory the driver can use.
executor_instance_typestrmx.smallThe Conveyor instance type to use for the executors. This specifies the CPU/Memory the executors can use.
executor_disk_sizeintThe total disk size attached to the executor nodes. Depending on the instance type the size will be divided over 1, 2 or 4 disks.
aws_rolestrThe AWS role to be used by both the driver and executors.
aws_availability_zonestrThe availability zone where the Spark application should run (e.g. eu-west-1a).
azure_application_client_idstringThe Azure service principal used by the container.
env_varsdictEnvironment variables for the spark driver (templated).
instance_life_cyclestringspotThe lifecycle of the instance used to run this job. Options are on-demand, spot or driver-on-demand-executors-spot.
s3_committerstrfileThe S3 committer to be used by the Spark applications. Supported committers are file and magic. The magic S3 committer requires a Spark installation with hadoop-cloud support. This is provided by our ...hadoop-3.3.1-...-v2 images and later.
abfs_committerstrfileThe ABFS committer to be used by the Spark applications. Supported committers are file and manifest. The manifest ABFS committer requires a Spark installation with hadoop-cloud support. This is provided by our ...hadoop-3.3.5-...-v1 images and later.
modestrclusterThis settings allows you to change the Spark mode between cluster, cluster-v2 and local. For more info see here
verboseboolFalseThis setting will configure spark-submit to run with the --verbose option, printing out fine-grained debugging information.

In case the aws_role or azure_application_client_id is not supplied to the operator, the default identity configured on the project will be used instead.

Setting CPU/Memory​

You can use the driver_instance_type and executor_instance_type to set the CPU/Memory of your spark driver and executors. We support this to make it easier to select CPU/Memory settings for your driver and executors you can set it to the mx.micro instance for example:

from conveyor.operators import ConveyorSparkSubmitOperatorV2

ConveyorSparkSubmitOperatorV2(
    ...,
    driver_instance_type='mx.micro',
    executor_instance_type='mx.micro',
)

When running Spark/PySpark applications, only a part of the total memory for the container is available for Spark itself. The details are described in the following tables:

AWS​

Instance typeCPUTotal Memory (AWS)Spark memory (AWS)PySpark memory (AWS)
mx.micro1*0.875 Gb0.8 Gb0.6 Gb
mx.small1*1.75 Gb1.6 Gb1.25 Gb
mx.medium13.5 Gb3.2 Gb2.5 Gb
mx.large27 Gb6.4 Gb5 Gb
mx.xlarge414 Gb12.7 Gb10 Gb
mx.2xlarge829 Gb26.7 Gb21 Gb
mx.4xlarge1659 Gb54 Gb42.4 Gb
cx.medium11.75 Gb1.6 Gb1.25 Gb
cx.large23.5 Gb3.2 Gb2.5 Gb
cx.xlarge47 Gb6.4 Gb5 Gb
cx.2xlarge814 Gb12.7 Gb10 Gb
cx.4xlarge1629 Gb26.7 Gb21 Gb
rx.xlarge828 Gb26 Gb21 Gb
rx.2xlarge1659 Gb54 Gb43 Gb
rx.4xlarge16120 Gb112 Gb88 Gb
info

(*) These instance types don't get a guaranteed full CPU but only a slice of a full CPU, but they are allowed to burst up to a full CPU if the cluster allows.

Azure​

Instance typeCPUTotal Memory (Azure)Spark memory (Azure)PySpark memory (Azure)
mx.micro1*0.75 Gb0.69 Gb0.55 Gb
mx.small1*1.5 Gb1.38 Gb1.1 Gb
mx.medium13 Gb2.75 Gb2.15 Gb
mx.large26 Gb5.5 Gb4.3 Gb
mx.xlarge412 Gb11 Gb8.6 Gb
mx.2xlarge826 Gb23.6 Gb18.6 Gb
mx.4xlarge1655 Gb50 Gb35.7 Gb
info

(*) These instance types don't get a guaranteed full CPU but only a slice of a full CPU, but they are allowed to burst up to a full CPU if the cluster allows.

As you can see from the tables, the supported executor memory configs change depending on using regular (Scala) Spark or PySpark. The explanation for this can be found in the spark.kubernetes.memoryOverheadFactor which can be found in the Spark settings. This setting is configured to 0.1 for JVM jobs (Scala and Java Spark), and to 0.4 for non-JVM jobs (PySpark, SparkR). A portion of the memory is set aside for non-JVM things like: off-heap memory allocations, system-processes, Python, R... Otherwise, your job would commonly fail with the error "Memory Overhead Exceeded".

Default Spark configuration​

A number of configuration options are set by default, but can be overwritten by passing using the conf dictionary:

ConfigurationDefault value
spark.kubernetes.container.image.pullPolicyAlways
spark.kubernetes.pyspark.pythonVersion3
spark.hadoop.mapreduce.fileoutputcommitter.algorithm.version2
spark.hadoop.hive.metastore.client.factory.class (spark 2.4)com.amazonaws.glue.catalog.metastore.AWSGlueDataCatalogHiveClientFactory
spark.hadoop.hive.imetastoreclient.factory.class (spark 3.0)com.amazonaws.glue.catalog.metastore.AWSGlueDataCatalogHiveClientFactory
spark.hadoop.fs.s3.implorg.apache.hadoop.fs.s3a.S3AFileSystem
spark.hadoop.fs.s3a.aws.credentials.providercom.amazonaws.auth.DefaultAWSCredentialsProviderChain
spark.executor.extraJavaOptions-Dlog4j.configuration=file:///opt/spark/log4j/log4j.properties -Dlog4j2.configurationFile=file:///opt/spark/log4j/log4j2.properties
spark.driver.extraJavaOptions-Dlog4j.configuration=file:///opt/spark/log4j/log4j-executor.properties -Dlog4j2.configurationFile=file:///opt/spark/log4j/log4j2-executor.properties
spark.decommission.enabledTrue (only if spark version >= 3.2.0 )
spark.decommission.killInterval120 (only if spark version >= 3.2.0 )
spark.decommission.forceKillTimeout180 (only if spark version >= 3.2.0 )
spark.storage.decommission.enabledTrue (only if spark version >= 3.2.0 )
spark.storage.decommission.shuffleBlocks.enabledTrue (only if spark version >= 3.2.0 )
spark.storage.decommission.rddBlocks.enabledTrue (only if spark version >= 3.2.0 )

Spark 3 Support​

Spark 3 introduced some changes on how Spark on Kubernetes works, including one breaking change. When supplying the path to the jar in the container, it should be prefixed with local:// to inform Spark that the jar is inside the docker container.

Spark 3 also introduced changes in the Hive/Glue support. The ConveyorSparkSubmitOperatorV2 handles this correctly, which means you do not need to pass the spark.hadoop.hive.metastore.client.factory.class configuration yourself.

An example can be found here:

from conveyor.operators import ConveyorSparkSubmitOperatorV2

role = "job-role-{{ macros.conveyor.env() }}"

ConveyorSparkSubmitOperatorV2(
    task_id="the-task-id",
    num_executors=1,
    driver_instance_type="mx.small",
    executor_instance_type="mx.small",
    aws_role=role,
    java_class="name-of-the-class",
    application="local://path-to-jar-in-container",
    application_args=[
        "--environment", "{{ macros.conveyor.env() }}",
        "--snapshotDates", "2019-11-06"
    ]
)

Instance life cycle​

An instance life cycle can be set on the operator. This setting controls whether your job will run on on-demand or on spot instances. Spot instances can result in discounts of up to 90% compared to on-demand prices. The downside is that your job can be canceled when AWS reclaims such a spot instance, which is what we call a spot interrupt. Fortunately, this does not happen frequently.

For Spark applications, we provide 3 options:

  • on-demand: All containers will be run on on-demand instances. This is useful if your need your spark job to always succeed within a certain time limit or you will miss an SLA. As losing an executor on a spot instance might make your job take longer.
  • spot: All containers will be run on spot instances. This is the cheapest method, however your driver being killed by a spot interruption will fail your whole job.
  • driver-on-demand-executors-spot: Your driver will be run on on-demand instances, but your spark executors will run on spot instances. This is a good compromise between robustness and cost efficiency, this should probably be your default for production unless you have a very strict SLA.

The Airflow executor that is running your job will follow the instance life cycle of the driver. Meaning if you choose instance life cycle spot, the Airflow executor will run on spot instances. If you use on-demand or driver-on-demand-executors-spot the executor will run on on-demand instances.

Example configuration using instance life cycle:

from conveyor.operators import ConveyorSparkSubmitOperatorV2

ConveyorSparkSubmitOperatorV2(
    ...,
    instance_life_cycle = 'driver-on-demand-executors-spot',
)

Environment variables​

On the operator we can set environment variables. This is a convenient and recommended approach to pass configuration that is likely to vary between deployments. A simple example of a static environment variable looks as follows:

ConveyorSparkSubmitOperatorV2(
    env_vars={
        "STATIC_KEY": "hello world",
    },
)

Sometimes these configurations contain sensitive information and should not be part of the operator configuration. To support these use cases, you can load environment variables dynamically from a secrets store. We support the following stores:

AWS secrets​

To access the secrets, the AWS IAM role configured in the operator needs to have the permissions to access the secret. The following snippet shows how to use an AWS secret in your operator:

from conveyor.secrets import AWSParameterStoreValue, AWSSecretsManagerValue

ConveyorSparkSubmitOperatorV2(
    env_vars={
        "USERNAME": AWSParameterStoreValue(name="/example/username"),
        "PASSWORD": AWSSecretsManagerValue(name="example-password"),
    },
    aws_role="role-with-access-to-secrets",
)

Both stores also support selecting properties from JSON stored secrets using jmesPath syntax. For example, when storing a secret as json:

{
    "username": "ADMIN",
    "password": "MYSECRETPASSWORD"
}

You can access the secret as follows:

from conveyor.secrets import AWSSecretsManagerValue

ConveyorSparkSubmitOperatorV2(
    env_vars={
        "USERNAME": AWSSecretsManagerValue(name="example", path="username"),
        "PASSWORD": AWSSecretsManagerValue(name="example", path="password"),
    },
    aws_role="role-with-access-to-secrets",
)

AWS Secrets Manager IAM access​

To be able to access a secret from AWS Secrets Manager, you need to add the following actions to your IAM role:

  • secretsmanager:GetSecretValue
  • secretsmanager:DescribeSecret

You should scope these actions to the resulting resource for example:

{
    "Version": "2012-10-17",
    "Statement": [{
        "Effect": "Allow",
        "Action": ["secretsmanager:GetSecretValue", "secretsmanager:DescribeSecret"],
        "Resource": ["arn:*:secretsmanager:*:*:secret:MySecret-??????"]
    }]
}

The reason for the 6 ?'s is that AWS always adds 6 random characters at the end of the ARN of a secret. For more info look at the AWS docs.

AWS Parameter Store IAM access​

To be able to get secrets from the AWS SSM Parameter Store, you need to add the following actions to your IAM role:

  • ssm:GetParameters
  • ssm:GetParametersByPath

You should scope these actions to the correct parameter, for example:

{
    "Version": "2012-10-17",
    "Statement": [{
        "Effect": "Allow",
        "Action": ["ssm:GetParameters", "ssm:GetParametersByPath"],
        "Resource": ["arn:*:ssm:*:*:parameter/my-parameter"]
    }]
}

For more info, you can refer to the AWS docs.

Azure secrets​

In order to access the secrets, you must give the Azure Client ID access to the Key Vault that contains your secrets. The following snippet shows how to use an Azure secret in your operator:

from conveyor.secrets import AzureKeyVaultValue

ConveyorSparkSubmitOperatorV2(
    env_vars={
        "PASSWORD": AzureKeyVaultValue(name="mySecretKey",vault="myKeyVault",vault_type="secret")
    },
    azure_application_client_id="azure-client-id-with-access-to-secrets",
)
important

The vault_type indicates which type of resource the value is in your keyvault. The following resource types exist: secret, key, certificate. The default value in the operator is secret and can thus be omitted. For more details, have a look at the Azure documentation

Azure Key Vault RBAC permissions​

To be able to access a secret from Azure keyvault, you need to give your Azure application client ID access to your keyvault. The recommended practice is to use Azure role based access control with keyvaults. You can then give your application access by assigning both of the following Azure roles:

  • Key Vault Reader: allows metadata operations on the keyvault but not reading sensitive values
  • Key Vault Secrets User: allows to read sensitive values in the keyvault

For more details take a look at the Azure documentation

An example of how to do this in terraform can look as follows

resource "azuread_application" "azure_application" {
  display_name = "azure-application"
}

resource "azuread_service_principal" "azure_application" {
  application_id               = azuread_application.azure_application.application_id
  app_role_assignment_required = false
}

resource "azurerm_role_assignment" "keyvault_read_project" {
  scope                = var.azure_keyvault_id
  role_definition_name = "Key Vault Reader"
  principal_id         = azuread_service_principal.azure_application.id
}

resource "azurerm_role_assignment" "keyvault_read_secret_project" {
  scope                = var.azure_keyvault_id
  role_definition_name = "Key Vault Secrets User"
  principal_id         = azuread_service_principal.azure_application.id
}

Executor disk size​

The variable executor_disk_size makes sure you have a certain disk size available on your executors. By default, the mounted disk space accessible by your Spark application is potentially shared with other containers, but also with Kubernetes. The machines where your applications run on come with a 100Gb disk attached by default. Your job can use a percentage of this volume.

By supplying the executor_disk_size setting, an extra volume exclusive to your container gets attached to the machine. This means you can be sure about both the size of the volume and its performance. For larger instances, more disks will be mounted automatically in order to increase performance.

On AWS, GP3 volumes are used. These volumes have 125MB/s throughput and 3000 IOPS by default, resulting in the following numbers.

Container instancenumber of disksexpected throughputexpected IOPS
mx.xlarge (or smaller)1125 MB/s3000
mx.x2large2250 MB/s6000
mx.x4large4500 MB/s12000

Mode​

from conveyor.operators import ConveyorSparkSubmitOperatorV2

ConveyorSparkSubmitOperatorV2(
    ...,
    mode='cluster',
)

There are currently 3 modes available for spark:

  • cluster: the default mode for spark jobs, it will launch a spark submitter pod, that will launch the driver and that will launch the executors.
  • cluster-v2: this mode used by default when running with conveyor run, and can be configured in the operator. It will directly launch the driver pod, cutting down on startup costs. It will also start placeholder executor pods that will boot up nodes immediately for your Spark job, once the driver launches the real spark executors these placeholder executors will be replaced. This mode will be max 3 minutes fast when new nodes will be launched on AWS.
  • local: this mode launches a single driver pod without executors, it is fast but only useful for small Spark jobs, but it the fastest way to launch spark jobs