问题标签 [data-ingestion]

我们计划使用 REST API 调用从端点摄取数据并将数据存储到 HDFS。REST 调用以定期方式（每天或每小时）完成。

我已经使用 Flume 完成了 Twitter 摄取，但我认为使用 Flume 不适合我当前的用例，因为我没有在 Twitter 中使用像这样的连续数据 Firehose，而是离散的常规时间限制调用。

我现在的想法是使用自定义 Java 来处理 REST API 调用并保存到 HDFS，然后在该 Java jar 上使用 Oozie 协调器。

我想听听有关设计以及用于此用例的基于 Hadoop 的组件的建议/替代方案（如果有比我现在想的更容易的话）。如果你觉得我可以坚持使用 Flume，那么请给我一个如何做到这一点的想法。

这是我第一次来这里，如果我发的不好，很抱歉，也很抱歉我的英语不好。

我正在尝试配置 Apache Flume 和 Elasticsearch 接收器。一切正常，似乎运行良好，但是启动代理时有2个警告；以下是：

我的代理配置：

它启动了netcat，一切都很好，但我担心这些警告，我不明白。

我在 python 中有一个数据框，其中包含我用于二进制分类的所有数据。我在两次迭代中摄取数据——一次是一个类的所有数据，然后是另一个类的所有数据。然后我对行进行随机化。我遇到的问题是每次重新运行脚本时，都会重新创建数据框的行并随机创建不可重现的结果。

我应该从外部文件运行数据框创建和随机化吗？在模型构建中是否有关于数据摄取的常见做法？

在这方面我没有尝试过任何尝试。我还想知道从统计的角度或惯例来看这样做是否有意义？我会尝试以下方法：

但是，每次我运行脚本时，它也会调用形成数据并将其随机化的外部脚本。所以这不是我正在寻找的解决方案。

我不能真正展示一个例子，我正在加载文档（文本文件） - 文档二进制分类。数据框的结构如下：

摄取代码：

我发现的所有搜索都显示了如何导入一个表或推荐 import-all-tables。如果我想从我的数据库中获取 440 个表中的 35 个怎么办？我可以只编写一个命令并用逗号分隔表，还是必须将其放入脚本中并一遍又一遍地复制和传递命令并每次更改表名？

我想做的事：

我担心我可能不得不做的事情：

这是我发布的第一个问题，如果我错过了一些信息和平庸的格式，我深表歉意。如果需要，我可以更新。

我将尝试添加尽可能多的细节。我有一个不太优化的 Spark 作业，它将 RDBMS 数据转换为 Neo4j 中的图形节点和关系。

去做这个。这是我遵循的步骤：

1. 使用 spark sql 和连接创建非规范化数据帧“数据”。

2. 'data' 中的 Foreach 行运行一个 graphInsert 函数，该函数执行以下操作：

   一个。读取
   b 行的内容。制定一个 Neo4j 密码查询（我们使用Merge命令，因此我们只有一个城市，例如在 Neo4j 中创建的芝加哥，而芝加哥将出现在 RDBMS 表中的多行中）
   c。连接到 neo4j
   d. 执行查询
   e. 断开与 Neo4j 的连接
   

这是我面临的问题列表。

1. 插入很慢。

我知道合并查询比创建慢，但有没有另一种方法可以做到这一点，而不是为每条记录连接和断开连接？这是我的第一个代码草案，也许我正在努力如何使用一个连接从不同 Spark 工作节点上的多个线程插入。因此，为每条记录连接和断开连接。

2. 该作业不可扩展。它仅在 1 个核心下运行良好。一旦我用 2 个 spark 核心运行作业，我就会突然得到 2 个同名的城市，即使我正在运行合并查询。例如，有 2 个芝加哥城市违反了 Merge 的使用。我假设合并功能类似于“如果不存在则创建”。

我不知道我的实现在 neo4j 部分或 spark 中是否错误。如果有人可以指导我查看任何有助于我以更好的规模实现这一点的文档，那将很有帮助，因为我有一个大的火花集群，我需要充分利用它来完成这项工作。

如果您有兴趣查看代码而不是算法。这是 scala 中的 graphInsert 实现：

如果我们在 sqoop 中使用 6 个 mapper 从 Oracle 导入数据，那么 sqoop 和 source 之间将建立多少个连接。

它是单个连接还是每个映射器有 6 个连接。

我有一个带有三个机架的集群。对于一组特定的源，我希望它们只被转储到一个机架中，以便我可以监控从该特定源到其他目的地的流量。我的问题很简单。是否可以将特定数据源复制到一个机架，而将其余数据源复制到所有其他（两个或三个）机架？

谢谢。

摘要：我有一个多路复用场景，并且想知道如何动态多路复用 - 不是基于静态配置的值，而是基于字段的变量值（例如日期）。

详细信息：我有一个输入，由 entityId 分隔。由于我知道我正在使用的实体，我可以在典型的 Flume 多通道选择中对其进行配置。

每个通道都转到一个单独的 HDFSEventSink，“hdfsSink-n”：

这会为每个实体生成一个文件，这很好。现在我要介绍第二个变量，它是动态的：日期。根据事件日期，我想创建文件 per-entity per-date。日期是一个动态值，因此我无法预先配置多个接收器，因此每个接收器都发送到一个单独的文件。此外，您只能为每个 Sink 指定一个 HDFS 输出。

所以，这就像是需要一个“Multiple Outputs HDFSEventSink”（与 Hadoop 的 MultipleOutputs 库类似）。Flume中有这样的功能吗？

如果没有，是否有任何优雅的方法来解决这个问题或解决这个问题？另一种选择是修改 HDFSEventSink ，它似乎可以通过为每个事件创建不同的“realName”（字符串）来实现。

我刚开始使用 Sqoop Hands-on。我有一个问题，假设我在数据库中有 300 个表，我想对这些表执行增量加载。我知道我可以使用附加模式或上次修改的方式进行增量导入。

但是，如果工作中唯一变化的是表名、CDC 列和最后一个值/更新值，我是否必须创建 300 个工作？

有没有人尝试使用相同的作业并将上述内容作为参数传递，该参数可以从循环中的文本文件中读取，并为所有表并行执行相同的作业。

行业标准和建议是什么？

此外，有没有办法截断并重新加载非常小的 hadoop 表，而不是执行 CDC 并稍后合并表？

当spark通过驱动程序提取数据时，我试图简单地说，然后当spark不需要通过驱动程序提取数据时。

我有3个问题-

1. 假设您有一个存储在 HDFS 中的 20 TB 平面文件文件，并从驱动程序中使用相应库的开箱即用功能之一（sc.textfile(path) 或sc.textfile(path).toDF等）将其拉入数据帧或 RDD。如果驱动程序仅使用 32 GB 内存运行，是否会导致驱动程序出现 OOM？或者至少有司机吉姆的交换？或者 spark 和 hadoop 是否足够聪明，可以将来自 HDFS 的数据分发到 spark 执行器中，从而在不通过驱动程序的情况下生成数据帧/RDD？
2. 除了来自外部 RDBMS 之外，与 1 完全相同的问题？
3. 与 1 完全相同的问题，除了来自特定节点文件系统（只是 Unix 文件系统，20 TB 文件但不是 HDFS）？