无法使用 Airflow 从 GCP Composer 连接到 FTP 服务器

Question

我在使用 google 的云应用程序（Airflow 在 Composer 和 Colaboratory 上运行）连接到 FTP 服务器时遇到了一些问题

在本地 Jupyter Notebook 上运行代码时，连接运行完美

from ftplib import FTP

ftp = FTP('ftp.mtps.gov.br')

ftp.login()

但在云应用程序上运行相同的行会返回

[2021-06-24 01:09:53,750] {taskinstance.py:1457} ERROR - [Errno 110] Connection timed out
Traceback (most recent call last):
  File "/opt/python3.8/lib/python3.8/site-packages/airflow/models/taskinstance.py", line 1113, in _run_raw_task
    self._prepare_and_execute_task_with_callbacks(context, task)
  File "/opt/python3.8/lib/python3.8/site-packages/airflow/models/taskinstance.py", line 1287, in _prepare_and_execute_task_with_callbacks
    result = self._execute_task(context, task_copy)
  File "/opt/python3.8/lib/python3.8/site-packages/airflow/models/taskinstance.py", line 1317, in _execute_task
    result = task_copy.execute(context=context)
  File "/opt/python3.8/lib/python3.8/site-packages/airflow/operators/python.py", line 117, in execute
    return_value = self.execute_callable()
  File "/opt/python3.8/lib/python3.8/site-packages/airflow/operators/python.py", line 128, in execute_callable
    return self.python_callable(*self.op_args, **self.op_kwargs)
  File "/home/airflow/gcs/dags/RAIS_ETL.py", line 31, in downloadData
    ftp = FTP('ftp.mtps.gov.br')
  File "/opt/python3.8/lib/python3.8/ftplib.py", line 117, in __init__
    self.connect(host)
  File "/opt/python3.8/lib/python3.8/ftplib.py", line 152, in connect
    self.sock = socket.create_connection((self.host, self.port), self.timeout,
  File "/opt/python3.8/lib/python3.8/socket.py", line 808, in create_connection
    raise err
  File "/opt/python3.8/lib/python3.8/socket.py", line 796, in create_connection
    sock.connect(sa)
TimeoutError: [Errno 110] Connection timed out

Answer 1

我使用示例代码，使用与您相同的导入功能，但将 URL 更改为ftp.us.debian.org 以连接到 FTP 服务器并尝试运行 dag 任务。

场景1：（方法中不传递任何参数）：

funcftp.py

def my_function():

   from ftplib import FTP

   ftp = FTP('ftp.us.debian.org')

   ftp.login()

dagftp.py

    from __future__ import print_function

    import datetime

    from airflow import models
    from airflow.operators import bash_operator
    from airflow.operators import python_operator

    from ftplib import FTP
    from funcftp import my_function
    YESTERDAY = datetime.datetime.now() - datetime.timedelta(days=1)


    default_dag_args = {
      # The start_date describes when a DAG is valid / can be run. Set this to a
      # fixed point in time rather than dynamically, since it is evaluated every
      # time a DAG is parsed. See:
      # https://airflow.apache.org/faq.html#what-s-the-deal-with-start-date
     'start_date':YESTERDAY,
}
# Define a DAG (directed acyclic graph) of tasks.
# Any task you create within the context manager is automatically added to the
# DAG object.
with models.DAG(
        'composer_ftp',
        schedule_interval=datetime.timedelta(days=1),
        default_args=default_dag_args) as dag:

# An instance of an operator is called a task. In this case, the
    # hello_python task calls the "greeting" Python function.
    hello_python = python_operator.PythonOperator(
        task_id='hello',
        python_callable=my_function)

    

    # Likewise, the goodbye_bash task calls a Bash script.
    goodbye_bash = bash_operator.BashOperator(
        task_id='bye',
        bash_command='echo Goodbye.')

    # Define the order in which the tasks complete by using the >> and <<
    # operators. In this example, hello_python executes before goodbye_bash.
    hello_python >> goodbye_bash

场景2（通过在方法中传递参数）

dag.py

from __future__ import print_function

import datetime

from airflow import models
from airflow.operators import bash_operator
from airflow.operators import python_operator

from ftplib import FTP
from myfun1 import my_function
YESTERDAY = datetime.datetime.now() - datetime.timedelta(days=1)

default_dag_args = {
   # The start_date describes when a DAG is valid / can be run. Set this to a
   # fixed point in time rather than dynamically, since it is evaluated every
   # time a DAG is parsed. See:
   # https://airflow.apache.org/faq.html#what-s-the-deal-with-start-date
   'start_date': YESTERDAY,
}
# Define a DAG (directed acyclic graph) of tasks.
# Any task you create within the context manager is automatically added to the
# DAG object.
with models.DAG(
     'demo_run',
     schedule_interval=datetime.timedelta(days=1),
     default_args=default_dag_args) as dag:

 # An instance of an operator is called a task. In this case, the
 # hello_python task calls the "greeting" Python function.
 hello_python = python_operator.PythonOperator(
     task_id='hello_world',
     python_callable=my_function,
     op_kwargs={"x" : "python"})

 # Likewise, the goodbye_bash task calls a Bash script.
 goodbye_bash = bash_operator.BashOperator(
     task_id='bye',
     bash_command='echo Goodbye.')


   # Define the order in which the tasks complete by using the >> and <<
   # operators. In this example, hello_python executes before goodbye_bash.

func.py

def my_function(x):

   from ftplib import FTP

   ftp = FTP('ftp.us.debian.org')

   ftp.login()
   return x + "  is a programming language"

1. 在其中一条错误消息中，它表明使用了引用对象的 self 关键字。如果您将所有方法都用作静态方法，则无需将 self 传递给该方法。因为静态方法可以在不创建对象的情况下调用，所以它们没有 self 关键字。

2. 如果您在方法中传递一些参数，则需要通过提供 op_args and_op_kwargs 参数来确保这些参数也传递给 DAG 任务。

3. 如果这不适合您，请提供您正在使用的代码。

Answer 2

这对您来说已经晚了大约 6 个月，但我今天遇到了这个确切的问题并遇到了这个问题，将我的解决方案留给可能遇到同样问题的其他人： 为了使连接正常工作，您需要将 Cloud NAT 与 Cloud Composer 一起设置，以便为工作人员提供公共互联网访问权限，如下所述：https://cloud.google.com/composer/docs/composer-2/private-ip-environments#public_internet_access_for_your_workflows