问题标签 [data-ingestion]

我设置了一个druid集群，OVERLORD和COORDINATER ui都可以工作。每个节点也可以启动并且没有错误。我通过快速启动批处理对其进行了测试，它有效。现在我想使用宁静来摄取kafka数据，当我运行“bin/tranquility kafka -configFile conf/kafka.json”命令时，结果如下：

但是数据无法保存 hdfs 并且 OVERLORD 和 COORDINATER ui 中没有任务，我如何定位错误并修复它们？

下面是我的集群：

经纪人：2个节点

协调器：2个节点

霸主：2个节点

历史：3个节点

中层管理器：3nodes

broker、Coordinator、Overlordad 设置在所有相同的 2 个节点上，Historical 和 Middle Manager 设置在所有相同的 3 个节点上。

我是 Apache Nifi 的新手

我们创建 Nifi 流，它使用来自 kafka 的 json 数据，结果在丰富后被发送到另一个 kafka 主题。但是 HBase 查找不返回键的值。相反，它返回键、值对，如 MapRecord[{SERIAL_NUM=123456789}]. 但我只需要'123456789'的值。

我无法解决这个问题。有人可以帮我解决这个话题吗？

我正在分享我的流程的输出。我们希望看到“hbase_integid”的值为“123456789”，但查找服务返回

MapRecord[{SERIAL_NUM=123456789}]"

我们如何仅提取 SERIAL_NUM 值（“ 123456789 ”）的值？

我使用 druid 索引服务来摄取 kafka 数据，当我创建一个数据源时，它运行良好。但是当我添加另一个数据源时，没有为第二个数据源分配任务。我应该如何使所有数据源都能正常工作？下面有 2 个数据源。

但是运行任务都是关于“inhouse_homepage”数据源的。工作容量（cores - 1）都是关于“inhouse_homepage”数据源的。请帮帮我。

我正在寻找验证超过 6 亿行和最多 30 列的 csv 文件的每一行（该解决方案必须处理该范围的几个大型 csv 文件）。

列可以是文本、日期或金额。csv 必须使用 40 条规则进行验证，一些规则会检查金额的正确性，其中一些会检查日期（格式）等......</p>

每个验证规则的结果必须保存并在之后显示。

验证数据后，将应用第二阶段的验证规则，此时间基于总和、平均值……每个规则的结果也必须保存。

我正在使用 Spark 加载文件。和

或者

要迭代每一行，我看到有两个选项：

• 使用dataset.map( row -> { something }); “Something”应该验证每一行并将结果保存在某处

但是由于“某事”块将在执行程序中执行，我不知道如何将其返回给驱动程序或将其存储在可以从驱动程序进程中检索到的某个位置。

• 第二个选项是使用dataset.collect: 但它会导致内存不足，因为所有数据都将加载到驱动程序中。我们可以使用“take”方法，然后从数据集中删除子集（使用过滤器）并重复操作，但我不喜欢这种方法

我想知道是否有人可以建议我一种解决此类问题的可靠方法。基本上保留 Spark 用于验证规则的第二阶段，并使用 Spark 或其他框架来摄取文件并执行并生成第一组验证规则

在此先感谢您的帮助

我正在尝试从 HANA 视图中进行 sqoop 导入。我尝试了很多方法，它仍然存在。任何人都有类似的经历，也请帮我弄清楚我是否遗漏了什么：

Sqoop 工作：

sqoop 导入 --driver com.sap.db.jdbc.Driver --connect 'jdbc:sap://hostname:30015?currentschema="_SYS_BIC"' --username HDP_READ --password-file /path/vangaphx/clrhanapwd - -query 'SELECT * from ZFI_LOS_SUMMARY_NET_GROSS WHERE $CONDITIONS' --delete-target-dir --target-dir /user/vangaphx/well2/SAPdatazone --num-mappers 1

错误：

我能够使用logstash将结构化数据导入elasticsearch并在Kibana中派生报告。我浏览了几篇文章。但是，我对如何使用摄取插件将文本（非结构化数据）导入弹性搜索没有清楚的了解？

我想根据条件设置配置单元变量。我不确定这在 HQL 中是否可行，如果可行，那么如何实现。

我想做这样的事情，

任何帮助和知识将不胜感激。

我正在使用 Apache NiFi 来摄取和预处理一些 CSV 文件，但是在长时间运行时，它总是失败。错误总是一样的：

在日志中搜索，我总是看到这个错误：

是什么让我相信根本原因是 Nifi 无法更新日志文件 ./flowfile_repository/journals/8772495.journal 因为该日志已经关闭**，如日志文件所示。

我该如何解决这个问题？

谢谢！

你好，先生，

在数据摄取模板中，我需要获取此属性，因为我有带有日期字段的数据

日期数据 12-07-2018 a 13-07-2018 b 14-07-2018 c 15-07-2018 d

在那，我想采取最新的，即 15-07-2018

如果日期字段有新数据 16-07-2018 e 那么我必须通过检查最后更新日期 15-07-2018 而不是从第一个 12-07-2018 检查来获得 16-07-2018

像这样，如果我得到 17-08-2108 f 那么必须通过检查最后一个新日期 16-07-2018 来得到 17-08-2018 ..

如何实现这一点，我必须在哪个处理器中进行修改或必须添加新属性 当提要再次运行时，它如何获取最新的水印并从那里工作

我有数据进入 spooldir，我正在使用水槽将其拾取并进一步转发以进行一些处理。

有些文件不是必需的，所以我在水槽中使用 igonorePattern 属性以避免被拾取。

但问题是，我收到的必需文件和非必需文件数量相同，而且我无法控制传入的数据，因此我必须接受进入 spooldir 的任何内容。

由于我有很多这些不需要的文件，我没有磁盘空间来长时间存储它们。因此，我想知道是否有办法让水槽自动删除这些文件，就像它对所有.COMPLETED文件一样（是的，我正在删除被水槽拾取的文件）