问题标签 [data-pipeline]

我使用 Spark 进行数据处理，但从数据源（主要是 csv 文件）开始，我想建立一个数据管道，它具有正确的阶段来控制/测试/操作数据并将它们部署到不同的“阶段”（CI- CD/QA/UAT/LIVE/等）。

是否有任何有效的数据管道“蓝图”？

我需要在 hadoop 中创建数据管道。我已经设置了数据导入、导出、脚本来清理数据，现在需要在管道中进行设置。

我一直在使用 Oozie 进行数据导入和导出计划，但现在还需要集成 R 脚本来进行数据清理过程。

我看到猎鹰也用于同样的用途。

1. 如何在cloudera中安装falcon？
2. 还有哪些其他工具可用于在 hadoop 中创建数据管道？

我创建了一个从 S3 提取数据并将其推送到 DynamoDB 的数据管道。
管道开始成功运行。
我将写入容量设置为20000 个单位，几个小时后写入量减少了一半，现在它仍在以3 个单位的写入容量运行。

（写入容量没有改变。管道从阈值开始，然后减少到 3 个单位并继续以这个速率运行）

减少的原因可能是什么？有没有办法让它更快？
谢谢。

可以通过 Data Pipeline 转储 DynamoDb，也可以在 DynamoDb 中导入数据。导入进展顺利，但数据始终附加到 DynamoDb 中已存在的数据。

现在，我找到了扫描 DynamoDb 并逐个删除项目或通过 Batch 删除项目的工作示例。但无论如何，对于大量数据来说，它并不是一个好的变体。

也可以完全删除表并创建它。但是随着那个变体索引将会丢失。

因此，最好的方法是通过 Data Pipeline 导入覆盖 DynamoDb 数据或以某种方式截断。有可能吗？如果是的话怎么可能？

要求：从 S3 备份位置恢复 DynamoDB 表。

我们创建了 Data Pipeline 作业，然后在 Architect Wizard 中编辑资源部分。

我们在核心实例计数下放置了 20 个实例，但在数据管道作业激活后，创建的 EMR 集群只有一个主实例和一个核心实例。

您能否建议我们，如何增加资源部分下的核心数量

我们使用 ruby​​ 脚本将数据从 MySQL 迁移到 Redshift(PostgreSQL)。目前我们使用 YAML 配置文件来维护模式信息（列名和类型）。所以每当 MySQL 表发生更改时，我们都需要手动更改 YAML 文件。现在，我们正在考虑将 YAML 配置作为 JSON 对象移动到数据库（PSQL）。但是我们仍然需要手动更改模式配置。我们希望自动化这个过程。（最好使用 Ruby。）

PS：我们使用 bin-log 来增量更新 AWS Redshift 中的数据。

我正在尝试使用 emr 将 json 数据从 s3 存储桶加载到 dynamodb。我成功上传了数据，但我的计数不匹配而没有引发任何错误。

为什么会这样？

我试图弄清楚如何配置/安排 Airflow DAG 在准确的时间每天运行两次，而不是在满足条件后同时运行两次。

我想在午夜和晚上 9 点运行相同的任务。

为此，我添加了一个cronto schedule_intervallike 0 0,21 * * *，因此它每天午夜和晚上 9 点运行。但是今天（4 月 27 日）的运行从昨天（4 月 26 日）的 00:00:00 开始，并且 00:00:00 和 21:00:00 运行同时运行。

预期行为将在今天（4 月 27 日）00:00:00 运行，21 小时后在 21:00:00 再次运行。

有任何想法吗？

最后，问题是：我怎样才能每天运行两次 DAG？

谢谢你。

不久前，我使用一个处于暂停状态的测试 DAG 设置了气流的测试安装。

现在，在这个系统运行了几个星期而实际上没有做太多事情（除了一些测试运行之外）之后，我想转储数据库并意识到它是巨大的。

这是由于 dag_pickle 表，气流显然会创建一个 pickle 对象并每秒将其保存在数据库中。

但是，数据库中的这些 pickle 行都与 dag 无关，dag 行中的pickle_id字段为空：

检查 dag_pickle 表，我们看到每秒都会创建一行：

（如您所见，这个表太大以至于它成为一个操作负担，它现在是 7.5 GB 大小）

我现在的问题是：

1. 所有这些泡菜行都用于任何用途，还是我可以保存删除它们？（我假设后者，因为没有来自 dag 的引用）是否有一些清理程序可以从数据库中删除未使用的内容？

2. 这是我的配置错误还是气流错误？我可以配置气流，以便仅在实际需要时创建这些泡菜对象吗？

3. donot_pickle配置选项到底有什么影响，如果 donot_pickle 设置为True ，我还能从另一台机器上的工作人员运行 DAG吗？

谢谢！

我有一个很好的直通工作管道，我在命令行上通过 luigi 运行的任务会触发所有必需的上游数据获取并以正确的顺序进行处理，直到它流入我的数据库。

好东西。不幸的是，第一次数据获取与远程 IMAP 服务器通信，每次获取都是一个新连接和一个新查询：非常慢。我知道如何在一个会话（任务实例）中获取所有单独的消息文件。我不明白如何让下游任务保持原样，一次处理一条消息，因为需要一条消息的任务会触发仅获取一条消息，而不是获取所有可用消息。我为错过明显的解决方案而提前道歉，但到目前为止，它让我很难过如何保持我漂亮的简单愚蠢的管道基本上保持原样，但让顶部的漏斗在一次调用中吸收所有数据。谢谢你的帮助。