airflow相关内容

即使安装了 Microsoft Visual C++ 14.0 也会出现 Pip 错误

我阅读了之前相关或问过的所有问题和答案,但仍然没有找到适合我的问题的答案. 我使用的是 python 3.6.5 并且 pip(和 setuptools)是最新的.我从这里安装了 Microsoft Visual C++ Redistributable for Visual Studio 2017:https://www.visualstudio.com/downloads/#build-t ..
发布时间:2021-12-02 16:38:04 其他开发

在 Apache Airflow DAG 中使用 AWS SES 发送失败的电子邮件

每当我的 DAG 中的任务无法运行或重试运行时,我都会尝试让 Airflow 使用 AWS SES 向我发送电子邮件.我也在使用我的 AWS SES 凭证,而不是我的通用 AWS 凭证. 我目前的airflow.cfg [email]email_backend =airflow.utils.email.send_email_smtp[smtp]# 如果您希望气流在重试、失败时发送电子邮件, ..
发布时间:2021-11-27 10:58:14 其他开发

为气流中的日志设置 s3

我正在使用 docker-compose 来设置可扩展的气流集群.我的方法基于这个 Dockerfile https://hub.docker.com/r/puckel/docker-airflow/ 我的问题是将日志设置为从 s3 写入/读取.当 dag 完成时,我收到这样的错误 *** 日志文件不是本地的.*** 在这里获取:http://ea43d4d49f35:8793/log/x ..
发布时间:2021-11-27 10:47:58 Python

使用 Airflow dag run 创建 EMR 集群,任务完成后,EMR 将终止

我有 Airflow 作业,它们在 EMR 集群上运行良好.我需要的是,假设我有 4 个需要 EMR 集群的气流作业,假设需要 20 分钟才能完成任务.为什么我们不能在 DAG 运行时创建 EMR 集群,一旦作业完成,它将终止创建的 EMR 集群. 解决方案 当然,这将是最有效地利用资源.让我警告你:这里面有很多细节;我会尽量列出尽可能多的内容.我鼓励您添加自己的综合答案,列出您遇到的任何 ..
发布时间:2021-11-27 10:08:44 其他开发

如何从 Airflow 向 EMR 集群提交 Spark 作业?

如何在 EMR 主集群(由 Terraform 创建)和 Airflow 之间建立连接.我在具有相同 SG、VPC 和子网的 AWS EC2 服务器下设置了气流. 我需要解决方案,以便 Airflow 可以与 EMR 对话并执行 Spark 提交. https://aws.amazon.com/blogs/big-data/build-a-concurrent-data-orchest ..
发布时间:2021-11-27 10:06:18 其他开发

无法通过 AWS 负载均衡器 HTTPS 访问气流 Web 服务器,因为气流将我重定向到 HTTP

我在 EC2 上配置了一个气流网络服务器,它侦听端口 8080. 我在 EC2 前面有一个 AWS ALB(应用程序负载均衡器),监听 https 80(面向互联网),实例目标端口面向 http 8080. 我无法浏览 https:// 来自浏览器,因为气流网络服务器将我重定向到 http :///admin,ALB 不监听. 如果我上网 https: ..
发布时间:2021-11-27 10:01:31 其他开发

使用 Airflow API 进行 JWT 身份验证

我正在尝试实施基于令牌的身份验证,作为触发气流 dag 的一部分.有没有办法添加 JWT 身份验证来生成 access_token 来触发 API?非常感谢任何帮助! 解决方案 我们的身份验证服务返回一个 JSON 响应,如下所示: {“clientToken":“322e8df6-0597-479e-984d-db6d8705ee66"} 这是我在气流 2.1 中使用 SimpleH ..
发布时间:2021-11-15 04:42:48 其他开发

使用 Airflow API 进行 JWT 身份验证

我正在尝试实施基于令牌的身份验证,作为触发气流 dag 的一部分.有没有办法添加 JWT 身份验证来生成 access_token 来触发 API?非常感谢任何帮助! 解决方案 我们的身份验证服务返回一个 JSON 响应,如下所示: {“clientToken":“322e8df6-0597-479e-984d-db6d8705ee66"} 这是我在气流 2.1 中使用 SimpleH ..
发布时间:2021-11-15 04:34:17 其他开发

使用 Airflow API 进行 JWT 身份验证

我正在尝试实施基于令牌的身份验证,作为触发气流 dag 的一部分.有没有办法添加 JWT 身份验证来生成 access_token 来触发 API?非常感谢任何帮助! 解决方案 我们的身份验证服务返回一个 JSON 响应,如下所示: {“clientToken":“322e8df6-0597-479e-984d-db6d8705ee66"} 这是我在气流 2.1 中使用 SimpleH ..
发布时间:2021-11-15 04:31:45 其他开发

使用 Dataflow 与 Cloud Composer

我想澄清一下 Cloud Dataflow 或 Cloud Composer 是否适合这项工作,但我从 Google 文档中不清楚. 目前,我正在使用 Cloud Dataflow 读取非标准 csv 文件——进行一些基本处理——并将其加载到 BigQuery 中. 让我举一个非常基本的例子: # file.csv输入\x01日期房子\x0112/27/1982汽车\x0111/9/ ..
发布时间:2021-11-11 22:31:37 其他开发

用于数据处理和作业调度的 Apache Airflow 或 Apache Beam

我正在努力提供有用的信息,但我远不是一名数据工程师. 我目前正在使用 python 库 pandas 对我的数据执行一系列转换,这些数据有很多输入(目前是 CSV 和 excel 文件).输出是几个excel文件.我希望能够使用并行计算执行计划的受监控批处理作业(我的意思是不像我对 Pandas 所做的那样顺序),每月一次. 我不太了解 Beam 或 Airflow,我很快通读了文档, ..
发布时间:2021-11-11 22:31:34 其他开发

Airflow 1.10.0 通过 Ansible

下面是我尝试安装 Airflow 1.10.0 的 Ansible 代码. sudo journalctl -uairflow-webserver -e 输出是 Dec 31 12:13:48 ip-10-136-94-232.eu-central-1.compute.internalairflow[22224]: ProgrammingError: (_mysql_exceptions. ..
发布时间:2021-11-11 01:30:57 其他开发

Airflow 1.10.0 通过 Ansible

下面是我尝试安装 Airflow 1.10.0 的 Ansible 代码. sudo journalctl -uairflow-webserver -e 输出是 Dec 31 12:13:48 ip-10-136-94-232.eu-central-1.compute.internalairflow[22224]: ProgrammingError: (_mysql_exceptions. ..
发布时间:2021-11-11 00:25:10 其他开发

如何使用 localstack s3 端点以编程方式设置 Airflow 1.10 日志记录?

为了将气流日志设置到 localstack s3 存储桶,对于本地和 kubernetes 开发环境,我正在关注 用于登录到 s3 的气流文档.为了提供一些背景信息,localstack 是一个本地 AWS 云堆栈,其中包括在本地运行的 s3 在内的 AWS 服务. 我在气流容器中添加了以下环境变量类似到另一个堆栈溢出帖子,以尝试登录到我的本地 s3 存储桶.这是我为所有气流容器添加到 do ..
发布时间:2021-10-27 19:03:29 其他开发

用于运行 shell 脚本的 Airflow Dataproc 操作符

是否有任何直接的方法可以将 shell 脚本运行到 dataproc 集群中.目前我可以通过 pysparkoperator 运行 shell(它调用另一个 python 文件,然后这个 python 文件调用 shell 脚本).我搜索了很多链接,但到目前为止还没有找到任何直接的方法. 如果有人能告诉我最简单的方法,那对我真的很有帮助. 解决方案 PIG job with sh o ..
发布时间:2021-10-26 18:05:01 其他开发

TemplateSyntaxError:在 205 Airflow 宏中出现意外字符 u'\\'

我收到以下错误: 文件“",第 1 行,在模板中TemplateSyntaxError:205 处的意外字符 u'\\' 当我将其包含在我的代码中时: '{{ macros.ds_format(macros.ds_add(ds, -13), "%Y-%m-%d", "%Y%m%d") }}' 我在vim中重新输入,以防编码错误,但仍然没有运气! 解决方案 改为: "{{ macr ..
发布时间:2021-10-26 18:04:58 其他开发

使用 Airflow 加载到 bigQuery 时如何忽略未知列?

我正在使用 GoogleCloudStorageToBigQueryOperator 将数据从 Google Storage 加载到 bigQuery Json 文件的列数可能比我定义的要多.在这种情况下,我希望加载作业继续 - 只需忽略这个无法识别的列.我尝试使用 ignore_unknown_values 参数,但没有任何区别. 我的接线员: def dc():返回 [{"name ..
发布时间:2021-10-26 18:04:55 其他开发

Airflow 远程日志记录连接 - Airflow 1.7.1.3

我已经看过下面的帖子,但我正在尝试为气流 1.7.1.3 设置这个 Airflow 远程日志记录不起作用 有没有人有连接中指定密钥文件以访问项目相关存储桶所需的格式的具体示例?{"项目":"","key_path":""}这是我试过的. 解决方案 您似乎在使用 Google 凭据时遇到了问题.有很多方法可以解决它,我将解释我的方法.您必须首先在 Airflow 中创建连接 ID ..
发布时间:2021-10-26 18:04:52 其他开发

运行过多任务时气流健康检查失败?

我有一个使用 LocalExecutor 在 Marathon 上运行的单个容器 Airflow 设置.我有一个运行状况检查,它 ping Airflow 网络服务器上的 /health 端点.它目前有 5 个 CPU 分配给它,并且网络服务器正在运行 4 个 Gunicorn.昨晚我有大约 25 个任务同时运行.这导致运行状况检查失败,没有有用的错误消息.容器刚刚收到一个 SIGTERM.我想知 ..
发布时间:2021-10-26 18:04:48 其他开发

如何修复 DAG 似乎丢失?

我想运行一个简单的 Dag“test_update_bq",但是当我去 localhost 我看到这个:DAG“test_update_bq"似乎丢失了.运行“airflow initdb"时没有错误,当我运行 test airflow test test_update_bq update_table_sql 2015-06-01 时,它已成功完成并且表已在 BQ 中更新.达格: from 气流 ..
发布时间:2021-10-26 18:04:45 其他开发