Apache Airflow FTP Provider

Warnings

• gotcha Incorrect Airflow connection configuration for FTP/FTPS. Users often use a generic 'HTTP' or 'SFTP' connection type, or fail to set the 'Conn Type' to 'FTP' for FTP/FTPS operations.
  Fix: In Airflow UI (Admin -> Connections), create a connection with 'Conn Type' explicitly set to 'FTP'. Ensure 'Host', 'Port', 'Login', and 'Password' fields are correctly filled. For FTPS, specify parameters like `explicit_ftp` and `ssl_context_kw` in the 'Extra' field as JSON.
• gotcha Distinction between FTP and FTPS, and explicit vs. implicit FTPS. The `FTPOperator` and `FTPHook` default to FTP. For FTPS, you must use `FTPSFileTransferOperator` or `FTPSHook` and configure the connection accordingly.
  Fix: For FTPS, use `FTPSFileTransferOperator` or `FTPSHook`. Configure the Airflow connection's 'Extra' field with JSON parameters like `{"explicit_ftp": true}` for explicit FTPS, or specific `ssl_context_kw` for advanced SSL settings. The port for FTPS is typically 21 for explicit or 990 for implicit.
• gotcha File transfer mode (binary vs. ASCII). Issues can arise when transferring text files in binary mode, or vice versa, especially across different operating systems with varying line endings.
  Fix: The `FTPOperator` and `FTPSFileTransferOperator` have a `transfer_mode` parameter (default is 'binary'). Explicitly set `transfer_mode='ascii'` when dealing with plain text files to ensure correct line ending handling during transfer.
• gotcha Large file transfers may encounter network timeouts, leading to task failures. The default FTP connection timeout might be too short for very large files or slow networks.
  Fix: Consider implementing retries for your Airflow tasks. For longer timeouts, you might need to specify a `timeout` parameter when initializing the Hook (e.g., `FTPHook(ftp_conn_id, timeout=300)`). If not directly exposed by the operator, you might need to extend the hook or pass it via the 'Extra' field of the Airflow connection if supported by the underlying library.

Install

• pip install apache-airflow-providers-ftp Install FTP Provider

Imports

• FTPOperator

  from apache_airflow_providers_ftp.operators.ftp import FTPOperator

• FTPHook

  from apache_airflow_providers_ftp.hooks.ftp import FTPHook

• FTPSFileTransferOperator

  from apache_airflow_providers_ftp.operators.ftps import FTPSFileTransferOperator

• FTPSHook

  from apache_airflow_providers_ftp.hooks.ftps import FTPSHook

Quickstart

This quickstart defines an Airflow DAG that demonstrates uploading a file to an FTP server using the `FTPOperator`. It first creates a local dummy file, then uploads it to the configured FTP server, and finally cleans up the local file. Remember to configure an 'FTP' type connection in the Airflow UI with `conn_id='ftp_default'` and appropriate host, login, and password.

from __future__ import annotations

import pendulum

from airflow.decorators import dag
from airflow.operators.bash import BashOperator
from apache_airflow_providers_ftp.operators.ftp import FTPOperator

# Ensure you have an FTP connection configured in Airflow UI (Admin -> Connections)
# with conn_id='ftp_default'.
# Set: Conn Id = ftp_default, Conn Type = FTP
# Host: your_ftp_host (e.g., 'localhost' or 'ftp.example.com')
# Port: 21 (or 22 for SFTP if using an SFTP provider, 990 for implicit FTPS)
# Login: your_username
# Password: your_password

@dag(
    dag_id="ftp_example_dag",
    start_date=pendulum.datetime(2023, 10, 26, tz="UTC"),
    catchup=False,
    schedule=None,
    tags=["ftp", "example", "provider"],
)
def ftp_dag():
    # Create a dummy local file to upload
    create_local_file = BashOperator(
        task_id="create_local_file",
        bash_command="echo 'Hello from Airflow FTP provider!' > /tmp/airflow_ftp_test.txt"
    )

    # Upload the file to FTP
    upload_file_to_ftp = FTPOperator(
        task_id="upload_file_to_ftp",
        ftp_conn_id="ftp_default",
        local_filepath="/tmp/airflow_ftp_test.txt",
        remote_filepath="/remote_airflow_test.txt",
        operation="put", # 'put' (upload), 'get' (download), 'delete' (delete remote)
        create_intermediate_dirs=True,
    )

    # Clean up the local dummy file
    clean_local_file = BashOperator(
        task_id="clean_local_file",
        bash_command="rm /tmp/airflow_ftp_test.txt"
    )

    create_local_file >> upload_file_to_ftp >> clean_local_file

ftp_dag()