Shulou(Shulou.com)06/03 Report--

@ [toc]

AirFlow FAQ installation problems 1. ERROR "python setup.py xxx" appears in the installation. Question:

First, you need to update the pip version by using 'pip install-- upgrade pip' command.

Second, the setuptools version is too old, so the following problems Command "python setup.py egg_info" failed with error code 1 in / tmp/pip-build-G9yO9Z/tldr/, also need you to update

File "/ tmp/pip-build-G9yO9Z/tldr/setuptools_scm-3.3.3-py2.7.egg/setuptools_scm/integration.py", line 9, in version_keywordFile "/ tmp/pip-build-G9yO9Z/tldr/setuptools_scm-3.3.3-py2.7.egg/setuptools_scm/version.py", line 66, in _ warn_if_setuptools_outdatedsetuptools_scm.version.SetuptoolsOutdatedWarning: your setuptools is too old (1) upgrade the pip version using the "pip install-upgrade pip" command. >   [xiaokang@localhost ~] $sudo pip install-- upgrade pip > (2) use the "pip install-- upgrade setuptools" command to upgrade the setuptools version. >   [xiaokang@localhost ~] $sudo pip install-- upgrade setuptools > after solving the above problems, you can successfully install the previously installed software # 2, ERROR: Cannot uninstall 'enum34'. # problem: ```python encountered the following error when installing Airflow: ERROR: Cannot uninstall 'enum34'. It is a distutils installed project and thus we cannot accurately determine which files belong to it which would lead to only a partial uninstall. Solution:

Sudo pip install-ignore-installed enum34

When there are other errors that cannot be upgraded, you can force the upgrade using the following command format:

   sudo pip install-- ignore-installed + module name

3. The installation software reported an error, indicating that the problem could not be found in the software package: ERROR: Command errored out with exit status 1:

ERROR: Command errored out with exit status 1: command: / usr/bin/python-c 'import sys, setuptools, tokenize; sys.argv [0] =' "'" / tmp/pip-install-oZ2zgF/flask-appbuilder/setup.py' "'"'; _ _ file__=' "'" / tmp/pip-install-oZ2zgF/flask-appbuilder/setup.py' "'"'; f=getattr (tokenize,'"'"'open'"' ", open) (_ _ file__) Code=f.read (). Replace ('\ r\ n','); f.close () Exec (compile (code, _ _ file__) '' exec' "') 'egg_info-- egg-base / tmp/pip-install-oZ2zgF/flask-appbuilder/pip-egg-info cwd: / tmp/pip-install-oZ2zgF/flask-appbuilder/ Complete output (3 lines): / usr/lib64/python2.7/distutils/dist.py:267: UserWarning: Unknown distribution option:' long_description_content_type' warnings.warn (msg) error in Flask- AppBuilder setup command: 'install_requires' must be a string or list of strings containing valid project/version requirement specifiers-- ERROR: Command errored out with exit status 1: python setup.py egg_info Check the logs for full command output. Solution:

Check the installation command, which is usually caused by an error that occurs because the installation package cannot be found.

4. Prompt that the file or directory Python.h cannot be found: src/spt_python.h:14:20: fatal error: Python.h: No such file or directory

ERROR: Command errored out with exit status 1: command: / usr/bin/python-u-c 'import sys, setuptools, tokenize; sys.argv [0] =' "'" / tmp/pip-install-YmiKzY/setproctitle/setup.py' "'"'; _ file__=' "'" / tmp/pip-install-YmiKzY/setproctitle/setup.py' "'"'; f=getattr (tokenize,'"'"'open'"' ", open) (_ _ file__) Code=f.read (). Replace ('\ r\ n','); f.close () Exec (compile (code, _ _ file__) '' exec' "') 'install--record / tmp/pip-record-XTav9_/install-record.txt-- single-version-externally-managed-- compile cwd: / tmp/pip-install-YmiKzY/setproctitle/ Complete output (15 lines): running install running build running build_ext building' setproctitle' extension creating build creating build/temp.linux-x86_64-2.7creating build/temp.linux-x86_64 -2.7/src gcc-pthread-fno-strict-aliasing-O2-g-pipe-Wall-Wp -D_FORTIFY_SOURCE=2-fexceptions-fstack-protector-strong-- param=ssp-buffer-size=4-grecord-gcc-switches-M64-mtune=generic-D_GNU_SOURCE-fPIC-fwrapv-DNDEBUG-O2-g-pipe-Wall-Wp -D_FORTIFY_SOURCE=2-fexceptions-fstack-protector-strong-- param=ssp-buffer-size=4-grecord-gcc-switches-M64-mtune=generic-D_GNU_SOURCE-fPIC-fwrapv-fPIC-DHAVE_SYS_PRCTL_H=1-DSPT_VERSION=1.1.10-I/usr/include/python2.7-c src/setproctitle.c-o build/temp.linux-x86_64-2.7/src/setproctitle.o In file included from src/spt.h:15:0 From src/setproctitle.c:14: src/spt_python.h:14:20: fatal error: Python.h: No such file or directory # include ^ compilation terminated. Error: command 'gcc' failed with exit status 1-- ERROR: Command errored out with exit status 1: / usr/bin/python-u-c' import sys, setuptools, tokenize; sys.argv [0] ='"/ tmp/pip-install-YmiKzY/setproctitle/setup.py'"'' _ _ file__=' "'/ tmp/pip-install-YmiKzY/setproctitle/setup.py'"''; f=getattr (tokenize, 'open' "'', open) (_ _ file__); code=f.read (). Replace (''\ r\ n'','\ n','\ n'); f.close () Exec (compile (code, _ file__,'"''exec'"' ")) 'install--record / tmp/pip-record-XTav9_/install-record.txt-- single-version-externally-managed-- compile Check the logs for full command output. Solution:

Due to the lack of python development package, yum install python-devel installation can solve the problem.

Dag problem 1. Bash_command='/root/touch.sh' execution command error. Question:

{taskinstance.py:1058} ERROR-bash / root/touch.shTraceback (most recent call last): File "/ usr/lib/python2.7/site-packages/airflow/models/taskinstance.py", line 915, in _ run_raw_task self.render_templates (context=context) File "/ usr/lib/python2.7/site-packages/airflow/models/taskinstance.py", line 1267 In render_templates self.task.render_template_fields (context) File "/ usr/lib/python2.7/site-packages/airflow/models/baseoperator.py", line 689, in render_template_fields self._do_render_template_fields (self, self.template_fields, context, jinja_env, set ()) File "/ usr/lib/python2.7/site-packages/airflow/models/baseoperator.py", line 696, in _ do_render_template_fields rendered_content = self.render_template (content) Context, jinja_env, seen_oids) File "/ usr/lib/python2.7/site-packages/airflow/models/baseoperator.py", line 723, in render_template return jinja_env.get_template (content). Render (* * context) File "/ usr/lib64/python2.7/site-packages/jinja2/environment.py", line 830, in get_template return self._load_template (name) Self.make_globals (globals)) File "/ usr/lib64/python2.7/site-packages/jinja2/environment.py", line 804, in _ load_template template = self.loader.load (self, name, globals) File "/ usr/lib64/python2.7/site-packages/jinja2/loaders.py", line 113, in load source, filename, uptodate = self.get_source (environment, name) File "/ usr/lib64/python2.7/site-packages/jinja2/loaders.py", line 187 In get_source raise TemplateNotFound (template) TemplateNotFound: bash / root/touch.sh solution: add an extra space after the executed command

  is a trap caused by airflow's use of jinja2 as a template engine. When using the bash command, a space must be added to the tail.

Airflow1, start worker Times error question: Running a worker with superuser privileges when theworker accepts messages serialized with pickle is a very bad ideatimes if you really want to continue then you have to set the C_FORCE_ROOTenvironment variable (but please think about this before you do).

Solution:

Add export clockwork row = "True" in / etc/profile

2. How does airflow batch unpause a large number of dag tasks

A small number of tasks can be started by ordering airflow unpause dag_id commands, or by clicking the start button in the web interface, but when there are too many tasks, it is troublesome to start them one by one. In fact, the dag information is stored in the database, and you can start dag tasks in batches by modifying the database information in batches. If you are using mysql as your sql_alchemy_conn, you only need to log in to the airflow database and update the is_paused field of the table dag to 0 to start the dag task.

Example: update dag set is_paused = 0 where dag_id like "benchmark%"

3. After executing a task, the scheduler process of airflow hangs and enters a state of suspended animation

The general reason for this is that the scheduler Scheduler generated the task but could not publish it. And there is no error message in the log.

The possible reason is that the Borker connection dependency library is not installed:

Execute pip install apache-airflow [redis] if redis is used as broker

Execute pip install apache-airflow [rabbitmq] if rabbitmq is used as broker

Also check to see if the scheduler node can access the rabbitmq properly.

4. When there are too many dag files defined, the scheduler node of airflow runs slowly.

The scheduler of airflow has two threads by default, which can be improved by modifying the configuration file airflow.cfg:

[scheduler] # The scheduler can run multiple threads in parallel to schedule dags.# This defines how many threads will run.# defaults to 2 where it is changed to 100max_threads = 1005, airflow log level change vi airflow.cfg [core] # logging_level = INFOlogging_level = WARNING

NOTSET

< DEBUG < INFO < WARNING < ERROR < CRITICAL 如果把log的级别设置为INFO， 那么小于INFO级别的日志都不输出， 大于等于INFO级别的日志都输出。也就是说，日志级别越高，打印的日志越不详细。默认日志级别为WARNING。 注意: 如果将logging_level改为WARNING或以上级别，则不仅仅是日志，命令行输出明细也会同样受到影响，也只会输出大于等于指定级别的信息，所以如果命令行输出信息不全且系统无错误日志输出，那么说明是日志级别过高导致的。 6、AirFlow: jinja2.exceptions.TemplateNotFound 这是由于airflow使用了jinja2作为模板引擎导致的一个陷阱，当使用bash命令的时候，尾部必须加一个空格: Described here : see below. You need to add a space after the script name in cases where you are directly calling a bash scripts in the bash_command attribute of BashOperator - this is because the Airflow tries to apply a Jinja template to it, which will fail.t2 = BashOperator(task_id='sleep',bash_command="/home/batcher/test.sh", // This fails with `Jinja template not found` error#bash_command="/home/batcher/test.sh ", // This works (has a space after)dag=dag)7、AirFlow: Task is not able to be run 任务执行一段时间后突然无法执行，后台worker日志显示如下提示: [2018-05-25 17:22:05,068] {jobs.py:2508} INFO - Task is not able to be run 查看任务对应的执行日志: cat /home/py/airflow-home/logs/testBashOperator/print_date/2018-05-25T00:00:00/6.log...[2018-05-25 17:22:05,067] {models.py:1190} INFO - Dependencies not met for ,dependency 'Task Instance State' FAILED: Task is in the 'success' state which is not a valid state for execution. The task must be cleared in order to be run. 根据错误提示，说明依赖任务状态失败，针对这种情况有两种解决办法: 使用airflow run运行task的时候指定忽略依赖task: $ airflow run -A dag_id task_id execution_date 使用命令airflow clear dag_id进行任务清理: $ airflow clear -u testBashOperator8、CELERY: PRECONDITION_FAILED - inequivalent arg 'x-expires' for queue 'celery@xxxx.celery.pidbox' in vhost ''在升级celery 4.x以后使用rabbitmq为broker运行任务抛出如下异常:[2018-06-29 09:32:14,622: CRITICAL/MainProcess] Unrecoverable error: PreconditionFailed(406, "PRECONDITION_FAILED - inequivalent arg 'x-expires' for queue 'celery@PQSZ-L01395.celery.pidbox' in vhost '/': received the value '10000' of type 'signedint' but current is none", (50, 10), 'Queue.declare')Traceback (most recent call last): File "c:\programdata\anaconda3\lib\site-packages\celery\worker\worker.py", line 205, in startself.blueprint.start(self)....... File "c:\programdata\anaconda3\lib\site-packages\amqp\channel.py", line 277, in _on_close reply_code, reply_text, (class_id, method_id), ChannelError,amqp.exceptions.PreconditionFailed: Queue.declare: (406) PRECONDITION_FAILED - inequivalent arg 'x-expires' for queue 'celery@PQSZ-L01395.celery.pidbox' in vhost '/': received the value '10000' of type 'signedint' but current is none 出现该错误的原因一般是因为rabbitmq的客户端和服务端参数不一致导致的，将其参数保持一致即可。 比如这里提示是x-expires 对应的celery中的配置是control_queue_expires。因此只需要在配置文件中加上control_queue_expires = None即可。 在celery 3.x中是没有这两项配置的，在4.x中必须保证这两项配置的一致性，不然就会抛出如上的异常。 我这里遇到的了两个rabbitmq的配置与celery配置的映射关系如下表： rabbitmqcelery4.xx-expirescontrol_queue_expiresx-message-ttlcontrol_queue_ttl9、CELERY: The AMQP result backend is scheduled for deprecation in version 4.0 and removal in version v5.0.Please use RPC backend or a persistent backend celery升级到4.x之后运行抛出如下异常: /anaconda/anaconda3/lib/python3.6/site-packages/celery/backends/amqp.py:67: CPendingDeprecationWarning: The AMQP result backend is scheduled for deprecation in version 4.0 and removal in version v5.0. Please use RPC backend or a persistent backend. alternative='Please use RPC backend or a persistent backend.') 原因解析: 在celery 4.0中 rabbitmq 配置result_backbend方式变了: 以前是跟broker一样:result_backend = 'amqp://guest:guest@localhost:5672//' 现在对应的是rpc配置:result_backend = 'rpc://' 参考链接:http://docs.celeryproject.org/en/latest/userguide/configuration.html#std:setting-event_queue_prefix 10、CELERY: ValueError('not enough values to unpack (expected 3, got 0)',) windows上运行celery 4.x抛出以下错误: [2018-07-02 10:54:17,516: ERROR/MainProcess] Task handler raised error: ValueError('not enough values to unpack (expected 3, got 0)',)Traceback (most recent call last): ...... tasks, accept, hostname = _locValueError: not enough values to unpack (expected 3, got 0) celery 4.x暂时不支持windows平台，如果为了调试目的的话，可以通过替换celery的线程池实现以达到在windows平台上运行的目的: pip install eventletcelery -A worker -l info -P eventlet 参考链接: https://stackoverflow.com/questions/45744992/celery-raises-valueerror-not-enough-values-to-unpack https://blog.csdn.net/qq_30242609/article/details/79047660 11、Airflow: ERROR - 'DisabledBackend' object has no attribute '_get_task_meta_for' airflow运行中抛出以下异常: Traceback (most recent call last):File "/anaconda/anaconda3/lib/python3.6/site-packages/airflow/executors/celery_executor.py", line 83, in sync......return self._maybe_set_cache(self.backend.get_task_meta(self.id))File "/anaconda/anaconda3/lib/python3.6/site-packages/celery/backends/base.py", line 307, in get_task_metameta = self._get_task_meta_for(task_id)AttributeError: 'DisabledBackend' object has no attribute '_get_task_meta_for'[2018-07-04 10:52:14,746] {celery_executor.py:101} ERROR - Error syncing the celery executor, ignoring it:[2018-07-04 10:52:14,746] {celery_executor.py:102} ERROR - 'DisabledBackend' object has no attribute '_get_task_meta_for' 这种错误有两种可能原因: CELERY_RESULT_BACKEND属性没有配置或者配置错误;celery版本太低，比如airflow 1.9.0要使用celery4.x，所以检查celery版本，保持版本兼容;12、airflow.exceptions.AirflowException dag_id could not be found xxxx. Either the dag did not exist or it failed to parse 查看worker日志airflow-worker.err airflow.exceptions.AirflowException: dag_id could not be found: bmhttp. Either the dag did not exist or it failed to parse.[2018-07-31 17:37:34,191: ERROR/ForkPoolWorker-6] Task airflow.executors.celery_executor.execute_command[181c78d0-242c-4265-aabe-11d04887f44a] raised unexpected: AirflowException('Celery command failed',)Traceback (most recent call last):File "/anaconda/anaconda3/lib/python3.6/site-packages/airflow/executors/celery_executor.py", line 52, in execute_commandsubprocess.check_call(command, shell=True)File "/anaconda/anaconda3/lib/python3.6/subprocess.py", line 291, in check_callraise CalledProcessError(retcode, cmd)subprocess.CalledProcessError: Command 'airflow run bmhttp get_op1 2018-07-26T06:28:00 --local -sd /home/ignite/airflow/dags/BenchMark01.py' returned non-zero exit status 1. 通过异常日志中的Command信息得知, 调度节点在生成任务消息的时候同时也指定了要执行的脚本的路径(通过ds参数指定)，也就是说调度节点(scheduler)和工作节点(worker)相应的dag脚本文件必须置于相同的路径下面，不然就会出现以上错误。 参考链接:https://stackoverflow.com/questions/43235130/airflow-dag-id-could-not-be-found 13、airlfow 的 REST API调用返回 Airflow 404 = lots of circles 出现这个错误的原因是因为URL中未提供origin参数，这个参数用于重定向，例如调用airflow的/run接口，可用示例如下所示: http://localhost:8080/admin/airflow/run?dag_id=example_hello_world_dag&task_id=sleep_task&execution_date=20180807&ignore_all_deps=true&origin=/admin 14、Broker与Executor选择 请务必使用RabbitMQ+CeleryExecutor, 毕竟这个也是Celery官方推荐的做法, 这样就可以使用一些很棒的功能, 比如webui上点击错误的Task然后ReRun 15、pkg_resources.DistributionNotFound: The 'setuptools==0.9.8' distribution was not found and is required by the application pip install distribution 16、Supervisor 在使用supervisor的启动worker,server,scheduler的时候, 请务必给配置的supervisor任务加上 environment=AIRFLOW_HOME=xxxxxxxxxx 主要原因在于如果你的supervisor是通过调用一个自定义的脚本来运行的, 在启动worker的时候会另外启动一个serve_log服务, 如果没有设置正确的环境变量, serve_log 会在默认的AIRFLOW_HOME里找日志, 导致无法在webui里查看日志 17、Serve_log 如果在多个机器上部署了worker, 那么你需要iptables开启那些机器的8793端口, 这样webui才能查看跨机器worker的任务日志 18、AMPQ库 celery提供了两种库来实现amqp, 一种是默认的kombu, 另外一个是librabbitmq, 后者是对其c模块的绑定, 在1.8.1版本中, 使用的kombu的时候会出现scheduler自动断掉的问题, 这个应该是其对应版本4.0.2的问题, 当切成librabbitmq的时候, server 与 scheduler运行正常, 但是worker的从来不consume任务, 最后查出原因: Celery4.0.2的协议发生了变化但是librabbitmq还没有对应修改, 解决方法是, 修改源码里的 executors/celery_executor.py文件然后加入参数 CELERY_TASK_PROTOCOL = 119、RabbitMQ连接卡死 运行一段时间过后, 由于网络问题导致所有任务都在queued状态, 除非把worker重启才能生效, 查资料有人说是clelery的broker pool有问题, 继续给celery_executor.py加入参数 BROKER_POOL_LIMIT=0 //不使用连接池 另外这样只会减少卡死的几率, 最好使用crontab定时重启worker 20、特定任务只在特殊机器上运行 可以给DAG中的task指定一个queue, 然后在特定的机器上运行 airflow worker -q=QUEUE_NAME 即可实现 21、RabbitMQ中的queue数量过多问题 celery为了让scheduler知道每个task的结果并且知道结果的时间为 O(1) , 那么唯一的解决方式就是给每一个任务创建一个UUID的queue, 默认这个queue的过期时间是1天, 可以通过更改celery_executor.py的参数来调节这个过期时间 CELERY_TASK_RESULT_EXPIRES = time in seconds22、airflow worker 角色不能使用根用户启动 原因：不能用根用户启动的根本原因，在于airflow的worker直接用的celery，而celery 源码中有参数默认不能使用ROOT启动，否则将报错 . C_FORCE_ROOT = os.environ.get('C_FORCE_ROOT', False)ROOT_DISALLOWED = """\Running a worker with superuser privileges when theworker accepts messages serialized with pickle is a very bad idea!If you really want to continue then you have to set the C_FORCE_ROOTenvironment variable (but please think about this before you do).User information: uid={uid} euid={euid} gid={gid} egid={egid}"""ROOT_DISCOURAGED = """\You're running the worker with superuser privileges: this isabsolutely not recommended!Please specify a different user using the --uid option.User information: uid={uid} euid={euid} gid={gid} egid={egid}""" 解决方案一：修改airlfow源码，在celery_executor.py中强制设置C_FORCE_ROOT from celery import Celery, platforms在app = Celery(…)后新增platforms.C_FORCE_ROOT = True重启即可 解决方案二：在容器初始化环境变量的时候，设置C_FORCE_ROOT参数，以零侵入的方式解决问题 强制celery worker运行采用root模式export C_FORCE_ROOT=True23、docker in docker 在dags中以docker方式调度任务时，为了container的轻量话，不做重型的docker pull等操作，我们利用了docker cs架构的设计理念，只需要将宿主机的/var/run/docker.sock文件挂载到容器目录下即可 docker in docker 资料 :https://link.zhihu.com/?target=http://wangbaiyuan.cn/docker-in-docker.html#prettyPhoto 24、多个worker节点进行调度反序列化dag执行的时候，报找不到module的错误 当时考虑到文件更新的一致性，采用所有worker统一执行master下发的序列化dag的方案，而不依赖worker节点上实际的dag文件，开启这一特性操作如下 worker节点上： airflow worker -cn=ip@ip -p //-p为开关参数，意思是以master序列化的dag作为执行文件，而不是本地dag目录中的文件master节点上： airflow scheduler -p 错误原因: 远程的worker节点上不存在实际的dag文件，反序列化的时候对于当时在dag中定义的函数或对象找不到module_name 解决方案一：在所有的worker节点上同时发布dags目录，缺点是dags一致性成问题 解决方案二：修改源码中序列化与反序列化的逻辑，主体思路还是替换掉不存在的module为main。修改如下: //models.py 文件，对 class DagPickle(Base) 定义修改import dillclass DagPickle(Base):id = Column(Integer, primary_key=True)# 修改前: pickle = Column(PickleType(pickler=dill))pickle = Column(LargeBinary)created_dttm = Column(UtcDateTime, default=timezone.utcnow)pickle_hash = Column(Text) tablename = "dag_pickle" def init(self, dag): self.dag_id = dag.dag_id if hasattr(dag, 'template_env'): dag.template_env = None self.pickle_hash = hash(dag) raw = dill.dumps(dag) 修改前: self.pickle = dag reg_str = 'unusualprefix\w*{0}'.format(dag.dag_id) result = re.sub(str.encode(reg_str), b'main', raw) self.pickle =result //cli.py 文件反序列化逻辑 run(args, dag=None) 函数 // 直接通过dill来反序列化二进制文件，而不是通过PickleType 的result_processor做中转 修改前: dag = dag_pickle.pickle 修改后：dag = dill.loads(dag_pickle.pickle) >

Solution 3: zero intrusive source code, re-create a function without module using python's types.FunctionType, so there will be no problem with serialization and deserialization

New_func = types.FunctionType ((lambda df: df.iloc [:, 0] .size = = xx) .code, {})

# 25. On the master node, the task log executed remotely cannot be viewed through webserver > reason: because airflow views the task execution log in master through the http service of each node, but the host_name stored in the task_ instance table is not ip, there is a problem with the way to obtain hostname. > solution: modify the get_hostname function in airflow/utils/net.py Add logic ````python//models.py TaskInstanceself.hostname = get_hostname () / net.py to get the hostname set in the environment variable first. Add a logic import osdef get_hostname () to get the environment variable in the get_hostname: "Fetch the hostname using the callable from the config or using`socket.getfqdn` as a fallback." # try to get the environment variable if 'AIRFLOW_HOST_NAME' in os.environ:return os.environ [' AIRFLOW_HOST " _ NAME'] # First we attempt to fetch the callable path from the config.try:callable_path = conf.get ('core' 'hostname_callable') except AirflowConfigException:callable_path = None# Then we handle the case when the config is missing or empty. This is the# default behavior.if not callable_path:return socket.getfqdn () # Since we have a callable path, we try to import and run it next.module_path, attr_name = callable_path.split (':') module = importlib.import_module (module_path) callable = getattr (module, attr_name) return callable ()

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.