Airflow FTP Manipulations.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Accessing Files on FTPs/SFTPs with Airflow.\n",
    "\n",
    "## Agenda:\n",
    "- Basic Airflow architecture.\n",
    "\n",
    "- Simple FTP requests\n",
    "\n",
    "- PythonOperator\n",
    "\n",
    "- Hook+Python Operator\n",
    "\n",
    "- Scheduling"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Airflow words.\n",
    "\n",
    "- DAG: Directed Acyclic Graph, the structure Airflow uses for its workflows. Each DAG has an ordering (one task can depend on another (Directed)) and contains no cycles (acyclic). A DAG is made up of seperate tasks that are the  configuration for the workflow's structure - all the heavy lifting is done in the hooks and operators.\n",
    "\n",
    "- Hooks: Files used by Airflow to interact with external systems (databases, APIs, etc.)\n",
    "\n",
    "- Operators: The atomic unit of logic in Airflow - these files determine how the work gets done."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Simple FTP requests:\n",
    "\n",
    "#### There are plenty of different modules for dealing with FTPs in Python. For SFTPs, _paramiko_ is the best library to use.\n",
    "#### This unsecured example uses ftplib.\n",
    "\n",
    "_Note_: For . All commands used here have a paramiko equivalent."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### _Suppose you have some workflow that downloads data off of an FTP, does some transformation, and uploads it on a cron schedule. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "from ftplib import FTP\n",
    "\n",
    "def download_file(connection, file_name):\n",
    "    \"\"\"\n",
    "    Downloads file from FTP.\n",
    "    \"\"\"\n",
    "\n",
    "    filename = 'sample_data.csv'\n",
    "\n",
    "    localfile = open(filename, 'wb')\n",
    "    ftp.retrbinary('RETR ' + filename, localfile.write, 1024)\n",
    "\n",
    "    ftp.quit()\n",
    "    localfile.close()\n",
    "    \n",
    "    #Do some manipulations to the local file. \n",
    "    \n",
    "\n",
    "def upload_file(connection, file_name):\n",
    "    \"\"\"\n",
    "    Uploads file as binary to FTP. \n",
    "    \"\"\"\n",
    "    filename = 'sample_data.csv'\n",
    "    ftp.storbinary('STOR '+filename, open(filename, 'rb'))\n",
    "    ftp.quit()\n",
    "    \n",
    "host = ''\n",
    "username = ''\n",
    "password = ''\n",
    "port = 21\n",
    "\n",
    "ftp = FTP(host)\n",
    "connection = ftp.login(username, password)\n",
    "\n",
    "\n",
    "download_file(connection, 'test.csv')\n",
    "upload_file(connection, 'test.csv')\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Doing this in Airflow.\n",
    "\n",
    "The fastest way to do this in a DAG is to use the PythonOperator:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# Import Airflow specific dependencies. \n",
    "from airflow import DAG\n",
    "from airflow.operators import DummyOperator\n",
    "from airflow.operators.python_operator import PythonOperator \n",
    "from datetime import datetime\n",
    "\n",
    "#Import the module for the FTP.. \n",
    "from ftplib import FTP\n",
    "\n",
    "\n",
    "#Define functions\n",
    "def upload_file(**kwargs):\n",
    "    \"\"\"\n",
    "    Uploads file as binary to FTP. \n",
    "    \"\"\"\n",
    "    \n",
    "    credentials = kwargs.get('templates_dict').get('credentials', None)\n",
    "    host = credentials['host']\n",
    "    username = credentials['username']\n",
    "    password = credentials['password']\n",
    "    \n",
    "    \n",
    "    ftp = FTP(host)\n",
    "    ftp.login(username, password)\n",
    "\n",
    "    \n",
    "    filename = 'sample_data.csv'\n",
    "    ftp.storbinary('STOR '+filename, open(filename, 'rb'))\n",
    "    ftp.quit()\n",
    "\n",
    "\n",
    "def download_file(**kwargs):\n",
    "    \"\"\"\n",
    "    Downloads file from FTP.\n",
    "    \"\"\"\n",
    "    \n",
    "    credentials = kwargs.get('templates_dict').get('credentials', None)\n",
    "    host = credentials['host']\n",
    "    username = credentials['username']\n",
    "    password = credentials['password']\n",
    "    \n",
    "\n",
    "    ftp = FTP(host)\n",
    "    ftp.login(username, password)\n",
    "\n",
    "    filename = 'sample_data.csv'\n",
    "\n",
    "    localfile = open(filename, 'wb')\n",
    "    ftp.retrbinary('RETR ' + filename, localfile.write, 1024)\n",
    "\n",
    "    ftp.quit()\n",
    "    localfile.close()\n",
    "\n",
    "\n",
    "default_args = {\n",
    "    'owner': 'airflow',\n",
    "    'start_date': datetime(2017, 12, 19)\n",
    "}\n",
    "\n",
    "\n",
    "# Schedule this DAG to run once.\n",
    "dag = DAG('test_ftp',\n",
    "          description='Manipulating FTPs with PythonOperators',\n",
    "          schedule_interval='@once',\n",
    "          start_date=datetime(2017, 12, 18),\n",
    "          default_args=default_args)\n",
    "\n",
    "# FTP creds\n",
    "credentials= {\n",
    "    'host' : ''\n",
    "    'username' : '',\n",
    "    'password' : '',\n",
    "    'port' : 21\n",
    "    \n",
    "}\n",
    "with dag:\n",
    "    # Dummy start DAG.\n",
    "    kick_off_dag = DummyOperator(task_id='kick_off_dag')\n",
    "\n",
    "    # Call the functions\n",
    "\n",
    "    download_file = PythonOperator(\n",
    "        task_id='download_file',\n",
    "        python_callable=download_file,\n",
    "        # This passes the date into the function as a dictionaryt.\n",
    "        templates_dict={'credentials': credentials},\n",
    "        provide_context=True\n",
    "    )\n",
    "    \n",
    "    upload_file = PythonOperator(\n",
    "    task_id='upload_file',\n",
    "    python_callable=upload_file, #function-name\n",
    "    # This passes the params into the function as a dictionaryt.\n",
    "    templates_dict={'credentials': credentials},\n",
    "    provide_context=True\n",
    "    )\n",
    "    \n",
    "    \n",
    "    # Set dependencies.First the kickoff, then the download, and finally, the upload.\n",
    "    # A task won't start until the one before it does.\n",
    "    # e.g. the upload won't start until the download taks finishes. \n",
    "    kick_off_dag >>  download_file >> upload_file"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Why use the PythonOperator?"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Airflow is made up of 3 core components: the webserver, the scheduler, and the executor.\n",
    "    \n",
    "    Webserver - Responsible for the UI in the browser.\n",
    "    Scheduler - Handles the scheduling and state of tasks.\n",
    "    Executor- Handles executing underlying code.\n",
    "    \n",
    "The scheduler \"heartbeats\" DAG files every few seconds before sending anything anything to the executor.\n",
    "Each \"heartbeat\" executes **all** top level code. Any code that isn't wrappped in an operator is executed \n",
    "each heartbeat, making it incredibly expensive.\n",
    "\n",
    "**Airflow Best Practice: Minimize top-level code. **\n",
    "    \n",
    "The PythonOperator is a quick and dirty way around this - just throw your function in a PythonOperator and you \n",
    "can leverage Airflow's scheduling and dependency capabilities. "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Lots of repeated code.\n",
    "\n",
    "Python Operators make it easy to take previous scripts and use Airflow for scheduling and manage depedencies, but they lead to a fair deal of repeated code and aren't anymore modular than regular python functions. Furthermore, they make the DAG file itself cluttered. \n",
    "\n",
    "** Airflow Best Practice: The DAG file should be as close to a \"config\" file as possible. **\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Using the FTP Hook.\n",
    "\n",
    "Using a hook to handle the connection can clean this code up a ton by handling the connection to the FTP.\n",
    "\n",
    "https://github.com/apache/incubator-airflow/blob/v1-8-stable/airflow/contrib/hooks/ftp_hook.py"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "from airflow import DAG\n",
    "from airflow.operators import DummyOperator\n",
    "from airflow.operators.python_operator import PythonOperator\n",
    "from datetime import datetime\n",
    "from airflow.contrib.hooks import FTPHook\n",
    "\n",
    "\n",
    "def upload_file(**kwargs):\n",
    "    \"\"\"\n",
    "    Uploads file as binary to FTP. \n",
    "    \"\"\"\n",
    "\n",
    "    hook = FTP(ftp_conn_id='astro_ftp').get_conn()\n",
    "\n",
    "    local_path = 'sample_data.csv'\n",
    "    remote_path = '/astro_test/saple_data.csv'\n",
    "    \n",
    "    hook.store_file(local_path, remote_path)\n",
    "    hook.close()\n",
    "\n",
    "\n",
    "def download_file(**kwargs):\n",
    "    \"\"\"\n",
    "    Downloads file from FTP.\n",
    "    \"\"\"\n",
    "    hook = FTPHook(ftp_conn_id='astro_ftp').get_conn()\n",
    "\n",
    "    local_path = 'sample_data.csv'\n",
    "    remote_path = '/astro_test/saple_data.csv'\n",
    "    \n",
    "    hook.retrieve_file(local_path, remote_path)\n",
    "    hook.close()\n",
    "\n",
    "\n",
    "default_args = {\n",
    "    'owner': 'airflow',\n",
    "    'start_date': datetime(2017, 12, 19)\n",
    "}\n",
    "\n",
    "\n",
    "# Schedule this DAG to run once.\n",
    "dag = DAG('test_ftp',\n",
    "          description='Manipulating FTPs with PythonOperators+Hooks',\n",
    "          schedule_interval='@once',\n",
    "          start_date=datetime(2017, 12, 18),\n",
    "          default_args=default_args)\n",
    "\n",
    "with dag:\n",
    "\n",
    "    kick_off_dag = DummyOperator(task_id='kick_off_dag')\n",
    "\n",
    "    upload_file = PythonOperator(\n",
    "        task_id='upload_file',\n",
    "        python_callable=upload_file,\n",
    "        # This passes the params into the function.\n",
    "        provide_context=True\n",
    "    )\n",
    "    \n",
    "    download_file = PythonOperator(\n",
    "        task_id='download_file',\n",
    "        python_callable=download_file,\n",
    "        # This passes the date into the function.\n",
    "        provide_context=True\n",
    "    )\n",
    "    \n",
    "    # Set dependencies.\n",
    "    kick_off_dag >> upload_file >> download_file"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Handling Connections.\n",
    "\n",
    "Notice when the hook was instanstiated, it was simply passed the name of the connection instead of the actual credentials used. All hooks inherit from the BaseHook, which has access to the Airflow database that stores connections."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "![connections](img/airflow_connections.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The Connections Panel can be accessed from the UI. from Admin -> Connections.\n",
    "\n",
    "Connections are fernet key encrypted after they're entered, prevent credentials from going into files, \n",
    "and can be used by other DAGs in the same instance."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Not \"Airflowic\" Enough\n",
    "\n",
    "Using the hook with the PythonOperator cut down on repeated code, \n",
    "but the DAG file doesn't read like a config file yet.\n",
    "\n",
    "To polish it off, we'll write a custom FTPtoS3Operator."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "from airflow.models import BaseOperator\n",
    "from airflow.hooks.S3_hook import S3Hook\n",
    "from airflow.contrib.hooks import FTPHook\n",
    "import os\n",
    "import pandas as pd\n",
    "import boa\n",
    "import logging\n",
    "\n",
    "\n",
    "class FTPToS3Operator(BaseOperator):\n",
    "    \"\"\"\n",
    "    SFTP To S3 Operator\n",
    "    :param ftp_conn_id:     The source FTP conn_id.\n",
    "    :type sftp_conn_id:     string\n",
    "    :param ftp_path:        The path to the file on the FTP client.\n",
    "    :type sftp_path:        string\n",
    "    :param s3_conn_id:      The s3 connnection id.\n",
    "    :type s3_conn_id:       string\n",
    "    :param s3_bucket:       The destination s3 bucket.\n",
    "    :type s3_bucket:        string\n",
    "    :param s3_key:          The destination s3 key.\n",
    "    :type s3_key:           string\n",
    "    \"\"\"\n",
    "\n",
    "    template_fields = ('s3_key',)\n",
    "\n",
    "    def __init__(self,\n",
    "                 ftp_conn_id,\n",
    "                 ftp_directory,\n",
    "                 local_path,\n",
    "                 s3_conn_id,\n",
    "                 s3_bucket,\n",
    "                 s3_key,\n",
    "                 *args,\n",
    "                 **kwargs):\n",
    "        super().__init__(*args, **kwargs)\n",
    "        self.ftp = ftp_conn_id\n",
    "        self.ftp_directory = ftp_directory\n",
    "        self.local_path = local_path\n",
    "        self.s3_conn_id = s3_conn_id\n",
    "        self.s3_bucket = s3_bucket\n",
    "        self.s3_key = s3_key\n",
    "\n",
    "    def execute(self, context):\n",
    "        \n",
    "        # In an operator, everything should be fed to the execute operator.\n",
    "\n",
    "        # Operators use hooks for external connections.\n",
    "        ftp = FTPHook(ftp_conn_id=self.ftp_conn_id)\n",
    "        s3_hook = S3Hook(self.s3_conn_id)\n",
    "        \n",
    "        # Log out everything in the directory. \n",
    "        logging.info(self.ftp_directory)\n",
    "        \n",
    "        download_file(self, ftp)\n",
    "        upload_to_s3(self, s3_hook)\n",
    "    \n",
    "    def download_file(self, ftp):\n",
    "        \n",
    "        ftp.retrieve_file(self.local_path, self.remote_path)\n",
    "        logging.info(\"Downloaded file!\")\n",
    "    \n",
    "    \n",
    "    def upload_to_s3(self, s3_hook):\n",
    "        s3_hook.load_file(\n",
    "            filename=(self.s3_key).split('/')[1],\n",
    "            key='{0}'.format(self.s3_key),\n",
    "            bucket_name=self.s3_bucket,\n",
    "            replace=True\n",
    "        )\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now that all the logic is offloaded to the operator, the DAG file itself is a lot cleaner:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# Import Airflow specific dependencies. \n",
    "from airflow import DAG\n",
    "from airflow.operators import DummyOperator\n",
    "from plugins.ftp_plugin.operators import FTPToS3Operator\n",
    "from datetime import datetime\n",
    "\n",
    "default_args = {\n",
    "    'owner': 'airflow',\n",
    "    'start_date': datetime(2017, 12, 19)\n",
    "}\n",
    "\n",
    "# Schedule this DAG to run once.\n",
    "dag = DAG('test_ftp',\n",
    "          description='FTPs with FTPOperator',\n",
    "          schedule_interval='@once',\n",
    "          start_date=datetime(2017, 12, 18),\n",
    "          default_args=default_args)\n",
    "\n",
    "\n",
    "with dag:\n",
    "    # Dummy start DAG.\n",
    "    kick_off_dag = DummyOperator(task_id='kick_off_dag')\n",
    "\n",
    "    # Call the functions\n",
    "\n",
    "    ftp_to_s3 = FTPToS3Operator(\n",
    "        task_id='download_file',\n",
    "        ftp_conn_id = 'astro-ftp',\n",
    "        ftp_director='',\n",
    "        local_path= '/temp/test_data.csv',\n",
    "        s3_conn_id='astronomer_s3',\n",
    "        s3_bucket ='astronomer-worflows-dev',\n",
    "        s3_key ='test_data.csv',\n",
    "    )\n",
    "    \n",
    "    # Set dependencies.First the kickoff, then the download, and finally, the upload.\n",
    "    # A task won't start until the one before it does.\n",
    "    # e.g. the upload won't start until the download taks finishes. \n",
    "    kick_off_dag >>  ftp_to_s3"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "All custom logic should go inside the operator class. \n",
    "For example, suppose we wanted to add an option to delete the file off of the FTP after downloading it:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "from airflow.models import BaseOperator\n",
    "from airflow.hooks.S3_hook import S3Hook\n",
    "from airflow.contrib.hooks import FTPHook\n",
    "import os\n",
    "import pandas as pd\n",
    "import boa\n",
    "import logging\n",
    "\n",
    "\n",
    "class FTPToS3Operator(BaseOperator):\n",
    "    \"\"\"\n",
    "    SFTP To S3 Operator\n",
    "    :param ftp_conn_id:     The source FTP conn_id.\n",
    "    :type sftp_conn_id:     string\n",
    "    :param ftp_path:        The path to the file on the FTP client.\n",
    "    :type sftp_path:        string\n",
    "    :param s3_conn_id:      The s3 connnection id.\n",
    "    :type s3_conn_id:       string\n",
    "    :param s3_bucket:       The destination s3 bucket.\n",
    "    :type s3_bucket:        string\n",
    "    :param s3_key:          The destination s3 key.\n",
    "    :type s3_key:           string\n",
    "    \"\"\"\n",
    "\n",
    "    template_fields = ('s3_key',)\n",
    "\n",
    "    def __init__(self,\n",
    "                 ftp_conn_id,\n",
    "                 ftp_directory,\n",
    "                 local_path,\n",
    "                 s3_conn_id,\n",
    "                 s3_bucket,\n",
    "                 s3_key,\n",
    "                 delete=False, # Add another property here\n",
    "                 *args,\n",
    "                 **kwargs):\n",
    "        super().__init__(*args, **kwargs)\n",
    "        self.ftp = ftp_conn_id\n",
    "        self.ftp_directory = ftp_directory\n",
    "        self.local_path = local_path\n",
    "        self.s3_conn_id = s3_conn_id\n",
    "        self.s3_bucket = s3_bucket\n",
    "        self.s3_key = s3_key\n",
    "        self.delete = delete\n",
    "\n",
    "    def execute(self, context):\n",
    "        \n",
    "        # In an operator, everything should be fed to the execute operator.\n",
    "\n",
    "        # Operators use hooks for external connections.\n",
    "        ftp = FTPHook(ftp_conn_id=self.ftp_conn_id)\n",
    "        s3_hook = S3Hook(self.s3_conn_id)\n",
    "        \n",
    "        # Log out everything in the directory. \n",
    "        logging.info(self.ftp_directory)\n",
    "        \n",
    "        download_file(self, ftp)\n",
    "        upload_to_s3(self, s3_hook)\n",
    "    \n",
    "    def download_file(self, ftp):\n",
    "        \n",
    "        ftp.retrieve_file(self.local_path, self.remote_path)\n",
    "        if self.delete:\n",
    "            ftp.delete_file(self.remote_path)\n",
    "            \n",
    "        logging.info(\"Downloaded file!\")\n",
    "        \n",
    "        # Maybe add some custom transforms here.\n",
    "    \n",
    "    \n",
    "    def upload_to_s3(self, s3_hook):\n",
    "        s3_hook.load_file(\n",
    "            filename=(self.s3_key).split('/')[1],\n",
    "            key='{0}'.format(self.s3_key),\n",
    "            bucket_name=self.s3_bucket,\n",
    "            replace=True\n",
    "        )\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This results in a one line change to the DAG file."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# Import Airflow specific dependencies. \n",
    "from airflow import DAG\n",
    "from airflow.operators import DummyOperator\n",
    "from plugins.ftp_plugin.operators import FTPToS3Operator\n",
    "from datetime import datetime\n",
    "\n",
    "default_args = {\n",
    "    'owner': 'airflow',\n",
    "    'start_date': datetime(2017, 12, 19)\n",
    "}\n",
    "\n",
    "# Schedule this DAG to run once.\n",
    "dag = DAG('test_ftp',\n",
    "          description='FTPs with FTPOperator',\n",
    "          schedule_interval='@once',\n",
    "          start_date=datetime(2017, 12, 18),\n",
    "          default_args=default_args)\n",
    "\n",
    "\n",
    "with dag:\n",
    "    # Dummy start DAG.\n",
    "    kick_off_dag = DummyOperator(task_id='kick_off_dag')\n",
    "\n",
    "    # Call the functions\n",
    "\n",
    "    ftp_to_s3 = FTPToS3Operator(\n",
    "        task_id='download_file',\n",
    "        ftp_conn_id = 'astro-ftp',\n",
    "        ftp_director='',\n",
    "        local_path= '/temp/test_data.csv',\n",
    "        s3_conn_id='astronomer_s3',\n",
    "        s3_bucket ='astronomer-worflows-dev',\n",
    "        s3_key ='test_data.csv',\n",
    "        delete=True\n",
    "    )\n",
    "    \n",
    "    # Set dependencies.First the kickoff, then the download, and finally, the upload.\n",
    "    # A task won't start until the one before it does.\n",
    "    # e.g. the upload won't start until the download taks finishes. \n",
    "    kick_off_dag >>  ftp_to_s3"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "With all the heavy lifting done in the hook and operator files, the DAG file itself can mirror a config file:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# Import Airflow specific dependencies. \n",
    "from airflow import DAG\n",
    "from airflow.operators import DummyOperator\n",
    "from plugins.ftp_plugin.operators import FTPToS3Operator\n",
    "from datetime import datetime\n",
    "\n",
    "default_args = {\n",
    "    'owner': 'airflow',\n",
    "    'start_date': datetime(2017, 12, 19)\n",
    "}\n",
    "\n",
    "# Schedule this DAG to run once.\n",
    "dag = DAG('test_ftp',\n",
    "          description='FTPs with FTPOperator',\n",
    "          schedule_interval='@once',\n",
    "          start_date=datetime(2017, 12, 18),\n",
    "          default_args=default_args)\n",
    "\n",
    "files = [\n",
    "    {\n",
    "        'name' : 'sample_one.csv',\n",
    "        'delete': True,\n",
    "    },\n",
    "    \n",
    "    {\n",
    "        'name': 'sample_two.csv',\n",
    "        'delete': False\n",
    "    },\n",
    "    \n",
    "    {\n",
    "        'name' : 'sample_three.csv',\n",
    "        'delete':True\n",
    "        \n",
    "    }\n",
    "]\n",
    "\n",
    "with dag:\n",
    "    # Dummy start DAG.\n",
    "    kick_off_dag = DummyOperator(task_id='kick_off_dag')\n",
    "    for file in files:\n",
    "\n",
    "        ftp_to_s3 = FTPToS3Operator(\n",
    "            task_id='download_file_{0}'.format(file['name']),\n",
    "            ftp_conn_id = 'astro-ftp',\n",
    "            ftp_director='',\n",
    "            local_path= '/temp/{0}'.format(file['name']),\n",
    "            s3_conn_id='astronomer-s3',\n",
    "            s3_bucket ='astronomer-worflows-dev',\n",
    "            s3_key =file['name'],\n",
    "            delete=file['delete']\n",
    "        )\n",
    "\n",
    "    # Set dependencies.First the kickoff, then the download, and finally, the upload.\n",
    "    # A task won't start until the one before it does.\n",
    "    # e.g. the upload won't start until the download taks finishes. \n",
    "    kick_off_dag >>  ftp_to_s3"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.6.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}