问题标签 [data-ingestion]

我有一个交易算法，我正在 zipline 上进行回测。我已经成功地从 csv 文件中提取了美国普通股捆绑包。展望未来，我想在每个交易日结束时对其进行连续回测。

因此，我想通过从 Interactive Brokers 下载每个美国股票的每日 OHLCV 价格附加到我现有的捆绑包中（我已经编写了一个执行此操作的 python 脚本）。

现在我的问题是：如何将每个资产的新一天数据行附加到我现有的滑索捆绑包中？

具体来说，我不想创建新的捆绑包。

我正在寻找将 Google Analytics 数据（以及历史数据）引入 Redshift 的选项。欢迎任何关于工具、API 的建议。我在网上搜索并发现 Stitch 作为 ETL 工具之一，如果您有，请帮助我更好地了解此选项和其他选项。

我们的项目需要实时读取和解码消息文本文件。第三方使用 MQ 设置接收文件。要求是从 MQ 服务器中提取文件，将其存储在我们的系统中，使用规范解码并将数据存储在数据库表中。

这是建议的流程：

消息文件(.txt) --> MQ 服务器 --> ?? --> 解码 --> 存储在数据库中

问题：

1. 第 3 方已经有一个 MQ 设置，他们准备将它推送给我们。我可以知道如何订阅他们的 MQ 系统来接收文件吗？是否可以使用 SFTP 接收文件，或者我应该有一个 MQ 设置在我身边？
2. 我们已经有一个 Python 源代码来解码文件，但不确定如何在文件推送到我们身边时实时进行端到端处理。

我在 S3 中有一个加密的文本文件。我想使用 AWS Glue 将此数据加载到 RDS Aurora。

我创建了一个爬虫来映射我的加密文本文件的架构，但爬虫没有创建任何表。可以肯定的是，我爬取了相同的文件，但在解密版本上并且它可以工作。

我的问题是：

1. AWS Glue 不能直接从 S3 摄取加密文件吗？
2. 如果是，请让我知道我错过了什么？
3. 如果不是，那么将加密数据摄取到 AWS Glue 中的最佳方式是什么？（我的数据高度敏感）

提前致谢。

背景/上下文

我看到了几乎无数关于如何使用 Spring Batch 处理多个文件的示例，但是它们中的每一个都有一个所有文件都被处理成的对象。因此，许多包含兼容数据的文件都被处理到单个目标目标中，例如数据库表。

我想构建一个导入过程，它将接收十个不同的文件并将它们映射到同一个数据库/模式中的十个不同的目标表。文件名每天也会以可预测/可编码的方式略有变化，但我想我能处理好。我认为 Spring 可以做到这一点（多对多数据映射），但这是我不知道如何做的最后一件事。Spring 的声明式结构在某些方面非常有用，但老实说，我不确定如何设置多个映射，而且由于应用程序确实没有程序部分可言，所以我不能真正使用任何形式的迭代. 我可以简单地为每个文件制作单独的 jar，并在控制台上编写迭代脚本，但这也会使日志记录和报告变得复杂......坦率地说，这听起来很老套

问题

我如何告诉 Spring Batch 以十种不同的方式处理十个不同文件中的每一个，并将它们的数据映射到同一个数据库中的十个不同表中？

例子：

• 文件Data_20190501_ABC_000.txt包含 4 列波浪号分隔的数据，需要映射到ABC_data6 列的表（两列是元数据）
• 文件Data_20190501_DEF_000.txt包含 12 列波浪号分隔的数据，需要映射到DEF_data14 列的表（两列是元数据）
• 文件Data_20190501_GHI_000.txt包含 10 列波浪号分隔的数据，需要映射到GHI_data12 列的表（两列是元数据）
• 等等...十个不同的文件和表格

我可以处理波浪号分隔，我认为我可以以编程方式处理文件名中的日期，并且可以在 db 触发器中处理其中一个字段。另一个元数据字段应该是文件名，但这肯定是一个不同的问题。

更新

根据我认为 Mahmoud Ben Hassine 的建议，我为每个文件/表对创建了一个单独的读取器、映射器和写入器，并尝试使用 start(step1)、next(step2)、build() 范式以格式添加它们下面基于Spring 文档中的配置和运行作业中的示例：

任何一个步骤都独立运行，但是一旦我将一个步骤添加为“下一步”，它只会执行第一个步骤，并在日志输出中生成“步骤已经完成或不可重新启动，因此无需执行任何操作”信息消息 - 其中我要离开这里吗？

我们正在开发 MarkLogic 9.0.8.2

我们正在 Azure 上设置 MarkLogic 集群（3 个 VM），并且根据故障转移设计，希望在 Azure Blob 中有 3 个森林（每个用于节点）。

我完成了设置，当开始摄取时，我发现文档分布在 3 个森林中，而不是全部存储在每个森林中。

例如

我摄取了 30000 条记录，每个森林包含 10000 条记录。

我需要的是拥有 30000 条记录的所有森林。

是否有任何配置（在数据库或森林级别）我需要实现这一点？

MarkLogic 设置如下

• 3 台主机

数据配置 - 每个主机上有 1 个主林 - 不同主机上的每个主机有 1 个副本

我们在 Azure 虚拟机上部署了 MarkLogic 集群（3 台主机）和故障转移）

我们正在使用 MarkLogic ContentPump (MLCP) 将数据摄取到 MarkLogic

这是我们已经实施的

• 在第一台主机上安装 Java
• 复制的 MLCP 工具
• 通过提供第一台服务器作为主机参数来摄取数据

现在我们得到了一批要更新回 MarkLogic 的 xml

通过故障转移实施，由于某种原因第一台主机不可用，所以当我尝试通过第二台主机摄取数据时，我开始收到错误，即记录被摄取到不同的主机中，因此无法从这里进行更新。

所以我想知道摄取过程要遵循的最佳实践

我有来自 Kafka 的具有以下结构的事件数据，我想在 Druid 中摄取

具体来说，我想将维度“活动”从逗号分隔的字符串转换为数组/多值维度，以便可以很好地对其进行过滤和分组。到目前为止，我的摄取如下

然而，这会导致广告系列被作为字符串摄取。我既找不到在 flattenSpec 中使用 jq 表达式从中生成数组的方法，也找不到可以用作 transformSpec 的字符串拆分表达式之类的东西。

有什么建议么？

我正在从 Azure Blob 摄取一个大型的简单 json 数据集，并将数据移动到一个名为“cities_stage”的“阶段”，其中 FILE_FORMAT = json 就像这样。

（以下是错误步骤“解析 JSON 时出错：未知关键字“Hurzuf”，位置 7。”）

然后，我将这个舞台位置转储到一个带有单个变体列的表格中，就像这样。我正在摄取的文件大于 16mb，因此我使用 type = json strip_outer_array = true 为每个对象创建单独的行

当我从 city_raw_source 中选择 * 时，每一行如下所示。

当我添加对“国家”或“名称”的引用时，问题就出现了。这是我的查询（我没有在这个查询中使用国家，但它产生了相同的结果）。

错误：

解析 JSON 时出错：未知关键字“Hurzuf”，位置 7。

如果我删除“src:name”，ID、Lat 和 Lon 都会按预期返回

任何帮助表示赞赏！

我创建了一个示例应用程序来将事件发送到事件中心，事件中心随后将数据发送到 azure data explorer 数据库。我可以看到事件中心中出现的事件，但即使在从事件中心创建数据引入之后，Azure 数据资源管理器数据库中也没有引入相同的事件。

我已经从事件中心删除并重新创建了数据资源管理器以及数据库和数据摄取，但没有成功。顺便说一句，我也做了元数据映射。

数据应该插入到 ADX 数据库表中，我应该能够查询数据，但是对表的查询返回一个空白。