abhishek_singh
08/12/2022, 7:53 AMemcp
08/13/2022, 2:41 PMabhishek_singh
08/13/2022, 7:18 PMsteve_clarke
08/17/2022, 12:00 AM# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership. The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License. You may obtain a copy of the License at
#
# <http://www.apache.org/licenses/LICENSE-2.0>
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
"""
This is an example dag for ECSOperator.
The task "meltano" runs `meltano` task in `<my_cluster_name>` cluster.
It overrides the command in the `Meltano` container.
"""
import datetime
import os
from airflow import DAG
from airflow.providers.amazon.aws.operators.ecs import ECSOperator
dag = DAG(
dag_id="Salesforce_Testing",
default_args={
"owner": "airflow",
"depends_on_past": False,
"email": ["<your email address for failures>"],
"email_on_failure": True,
"email_on_retry": True,
},
default_view="graph",
schedule_interval=None,
start_date=datetime.datetime(2020, 1, 1),
tags=["example"],
)
# generate dag documentation
dag.doc_md = __doc__
# [START howto_operator_ecs]
airflow_logging_test = ECSOperator(
task_id="Meltano_Salesforce_to_Target",
dag=dag,
aws_conn_id="aws_ecs",
cluster="<my_cluster_name>",
task_definition="<my_task_definition_name>",
launch_type="FARGATE",
region_name="<insert aws region>",
overrides={
"containerOverrides": [
{
"name": "meltano",
"command": ["chamber", "exec", "meltano/tap-salesforce", "meltano/target-snowflake", "--", "meltano", "elt", "tap-salesforce", "target-snowflake", "state-id=tap_salesforce_to_target_snowflake"],
},
],
},
network_configuration={
"awsvpcConfiguration": {
"securityGroups": [os.environ.get("SECURITY_GROUP_ID", "mymeltano-containersecuritygroup")],
"subnets": [
os.environ.get("AIRFLOW__CORE__VPC_SUBNETA"),
os.environ.get("AIRFLOW__CORE__VPC_SUBNETB")
],
"assignPublicIp": "DISABLED"
},
},
tags={
"Customer": "Customer XYZ",
"Project": "Salesforce to Snowflake",
"Application": "Meltano",
"Version": "0.0.1",
"Environment": "Development",
},
awslogs_group="/ecs/meltanologs",
awslogs_region="<insert aws region>",
awslogs_stream_prefix="ecs/meltano", # prefix with container name
)
# [END howto_operator_ecs]