问题标签 [kite-sdk]

我有一个 Spark 作业，它从 HDFS 读取数百万条记录，处理它们，然后以 AVRO 格式写回 HDFS。观察到许多文件（写入）仍处于 .avro.tmp 状态。

我正在使用 Kite SDK 以 AVRO 格式写入数据。环境是CDH 5.5。

可能是因为 Spark 作业在完成读取记录并将它们发送给执行程序（实际上是在写入吗？）

如果是这种情况，我如何确保在所有 .tmp 都转换为 .avro 之前作业不会终止？或者还有什么可能的原因？

我是 NIFI 的新手，我的用例是从端口读取并以 parquet 格式写入 hdfs，我的研究表明有一种叫做 KiteSDK 的东西，我可以用它保存为 Parquet 格式。对吗？请指教。任何例子都会有所帮助。

在Ubuntu-14.04上使用HDP-2.5，运行此命令并

尝试import raw csv使用 KiteSdk 将数据输入 Hivever.1-1-0 并具有以下IOError：

1 个作业失败发生：org.kitesdk.tools.CopyTask: Kite(dataset:file:/tmp/444e6fc4-10e2-407d-afaf-723c408a6d... ID=1 (1/1)(1): java .io.FileNotFoundException：文件文件：/hdp/apps/2.5.0.0-1245/mapreduce/mapreduce.tar.gz 在 org.apache.hadoop.fs.RawLocalFileSystem.deprecatedGetFileStatus(RawLocalFileSystem.java:624) 中不存在.apache.hadoop.fs.RawLocalFileSystem.getFileLinkStatusInternal(RawLocalFileSystem.java:850) 在 org.apache.hadoop.fs.RawLocalFileSystem.getFileStatus(RawLocalFileSystem.java:614) 在 org.apache.hadoop.fs.DelegateToFileSystem.getFileStatus(DelegateToFileSystem) .java:125) 在 org.apache.hadoop.fs.AbstractFileSystem.resolvePath(AbstractFileSystem.java:468) 在 org.apache.hadoop.fs.FilterFs.resolvePath(FilterFs.java:158) 在 org.apache.hadoop。 fs。FileContext$25.next(FileContext.java:2195) at org.apache.hadoop.fs.FileContext$25.next(FileContext.java:2191) at org.apache.hadoop.fs.FSLinkResolver.resolve(FSLinkResolver.java:90)在 org.apache.hadoop.fs.FileContext.resolve(FileContext.java:2191) 在 org.apache.hadoop.fs.FileContext.resolvePath(FileContext.java:603) 在 org.apache.hadoop.mapreduce.JobSubmitter.addMRFrameworkToDistributedCache (JobSubmitter.java:457) 在 org.apache.hadoop.mapreduce.JobSubmitter.submitJobInternal(JobSubmitter.java:142) 在 org.apache.hadoop.mapreduce.Job$10.run(Job.java:1290) 在 org.apache .hadoop.mapreduce.Job$10.run(Job.java:1287) 在 java.security.AccessController.doPrivileged(Native Method) 在 javax.security.auth.Subject.doAs(Subject.java:422) 在 org.apache。hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1724) 在 org.apache.hadoop.mapreduce.Job.submit(Job.java:1287) 在 org.apache.crunch.hadoop.mapreduce.lib.jobcontrol.CrunchControlledJob。提交（CrunchControlledJob.java:329）在 org.apache.crunch.hadoop.mapreduce.lib.jobcontrol.CrunchJobControl.startReadyJobs（CrunchJobControl.java:204）在 org.apache.crunch.hadoop.mapreduce.lib.jobcontrol.CrunchJobControl。 pollJobStatusAndStartNewOnes(CrunchJobControl.java:238) at org.apache.crunch.impl.mr.exec.MRExecutor.monitorLoop(MRExecutor.java:112) at org.apache.crunch.impl.mr.exec.MRExecutor.access$000(MRExecutor .java:55) 在 org.apache.crunch.impl.mr.exec.MRExecutor$1.run(MRExecutor.java:83) 在 java.lang.Thread.run(Thread.java:745)doAs(UserGroupInformation.java:1724) at org.apache.hadoop.mapreduce.Job.submit(Job.java:1287) at org.apache.crunch.hadoop.mapreduce.lib.jobcontrol.CrunchControlledJob.submit(Cru​​nchControlledJob.java: 329）在 org.apache.crunch.hadoop.mapreduce.lib.jobcontrol.CrunchJobControl.startReadyJobs(CrunchJobControl.java:204) 在 org.apache.crunch.hadoop.mapreduce.lib.jobcontrol.CrunchJobControl.pollJobStatusAndStartNewOnes(CrunchJobControl.java: 238) 在 org.apache.crunch.impl.mr.exec.MRExecutor.monitorLoop(MRExecutor.java:112) 在 org.apache.crunch.impl.mr.exec.MRExecutor.access$000(MRExecutor.java:55) 在org.apache.crunch.impl.mr.exec.MRExecutor$1.run(MRExecutor.java:83) at java.lang.Thread.run(Thread.java:745)doAs(UserGroupInformation.java:1724) at org.apache.hadoop.mapreduce.Job.submit(Job.java:1287) at org.apache.crunch.hadoop.mapreduce.lib.jobcontrol.CrunchControlledJob.submit(Cru​​nchControlledJob.java: 329）在 org.apache.crunch.hadoop.mapreduce.lib.jobcontrol.CrunchJobControl.startReadyJobs(CrunchJobControl.java:204) 在 org.apache.crunch.hadoop.mapreduce.lib.jobcontrol.CrunchJobControl.pollJobStatusAndStartNewOnes(CrunchJobControl.java: 238) 在 org.apache.crunch.impl.mr.exec.MRExecutor.monitorLoop(MRExecutor.java:112) 在 org.apache.crunch.impl.mr.exec.MRExecutor.access$000(MRExecutor.java:55) 在org.apache.crunch.impl.mr.exec.MRExecutor$1.run(MRExecutor.java:83) at java.lang.Thread.run(Thread.java:745)mapreduce.Job.submit(Job.java:1287) 在 org.apache.crunch.hadoop.mapreduce.lib.jobcontrol.CrunchControlledJob.submit(Cru​​nchControlledJob.java:329) 在 org.apache.crunch.hadoop.mapreduce.lib。 jobcontrol.CrunchJobControl.startReadyJobs(CrunchJobControl.java:204) 在 org.apache.crunch.hadoop.mapreduce.lib.jobcontrol.CrunchJobControl.pollJobStatusAndStartNewOnes(CrunchJobControl.java:238) 在 org.apache.crunch.impl.mr.exec。 org.apache.crunch.impl.mr.exec.MRExecutor.access$000(MRExecutor.java:55) 在 org.apache.crunch.impl.mr.exec.MRExecutor$1 上的 MRExecutor.monitorLoop(MRExecutor.java:112)。在 java.lang.Thread.run(Thread.java:745) 处运行（MRExecutor.java:83）mapreduce.Job.submit(Job.java:1287) 在 org.apache.crunch.hadoop.mapreduce.lib.jobcontrol.CrunchControlledJob.submit(Cru​​nchControlledJob.java:329) 在 org.apache.crunch.hadoop.mapreduce.lib。 jobcontrol.CrunchJobControl.startReadyJobs(CrunchJobControl.java:204) 在 org.apache.crunch.hadoop.mapreduce.lib.jobcontrol.CrunchJobControl.pollJobStatusAndStartNewOnes(CrunchJobControl.java:238) 在 org.apache.crunch.impl.mr.exec。 org.apache.crunch.impl.mr.exec.MRExecutor.access$000(MRExecutor.java:55) 在 org.apache.crunch.impl.mr.exec.MRExecutor$1 上的 MRExecutor.monitorLoop(MRExecutor.java:112)。在 java.lang.Thread.run(Thread.java:745) 处运行（MRExecutor.java:83）java:329) at org.apache.crunch.hadoop.mapreduce.lib.jobcontrol.CrunchJobControl.startReadyJobs(CrunchJobControl.java:204) at org.apache.crunch.hadoop.mapreduce.lib.jobcontrol.CrunchJobControl.pollJobStatusAndStartNewOnes(CrunchJobControl. java:238) 在 org.apache.crunch.impl.mr.exec.MRExecutor.monitorLoop(MRExecutor.java:112) 在 org.apache.crunch.impl.mr.exec.MRExecutor.access$000(MRExecutor.java:55 ) 在 org.apache.crunch.impl.mr.exec.MRExecutor$1.run(MRExecutor.java:83) 在 java.lang.Thread.run(Thread.java:745)java:329) at org.apache.crunch.hadoop.mapreduce.lib.jobcontrol.CrunchJobControl.startReadyJobs(CrunchJobControl.java:204) at org.apache.crunch.hadoop.mapreduce.lib.jobcontrol.CrunchJobControl.pollJobStatusAndStartNewOnes(CrunchJobControl. java:238) 在 org.apache.crunch.impl.mr.exec.MRExecutor.monitorLoop(MRExecutor.java:112) 在 org.apache.crunch.impl.mr.exec.MRExecutor.access$000(MRExecutor.java:55 ) 在 org.apache.crunch.impl.mr.exec.MRExecutor$1.run(MRExecutor.java:83) 在 java.lang.Thread.run(Thread.java:745)apache.crunch.impl.mr.exec.MRExecutor.monitorLoop(MRExecutor.java:112) at org.apache.crunch.impl.mr.exec.MRExecutor.access$000(MRExecutor.java:55) at org.apache.crunch .impl.mr.exec.MRExecutor$1.run(MRExecutor.java:83) 在 java.lang.Thread.run(Thread.java:745)apache.crunch.impl.mr.exec.MRExecutor.monitorLoop(MRExecutor.java:112) at org.apache.crunch.impl.mr.exec.MRExecutor.access$000(MRExecutor.java:55) at org.apache.crunch .impl.mr.exec.MRExecutor$1.run(MRExecutor.java:83) 在 java.lang.Thread.run(Thread.java:745)

我已经检查了该文件是否"hdfs:/hdp/apps/2.5.0.0-1245/mapreduce/mapreduce.tar.gz" 存在，但很长一段时间都无法弄清楚如何解决此错误。

任何帮助是极大的赞赏。

在处理水槽（1.6 和 1.7）时，我遇到了以下错误

我的代理管道设置为从 spooldir 源摄取 json 数据，使用 morphline 拦截器提取和转换。

请参阅下面的水槽配置摘录

我在 NiFi 中建立了一个管道，我在其中获取 JSON 记录，然后我使用这些记录向 API 发出请求。我得到的响应将包含数字和文本数据。然后我必须将此数据写入 Hive。我InferAvroSchema用来推断模式。一些数值是带符号的值，例如-2.46,-0.1在推断类型时，处理器将它们视为string而不是double或float或decimal类型。

我知道我们可以在处理器中对我们的 AVRO 模式进行硬编码，但我认为利用 AVRO 模式使其更具动态性InferAvroSchema会更好。还有其他方法可以克服/解决这个问题吗？

我正在使用该类编写 Parquet 文件ParquetDatasetStoreWriter，但我得到的性能非常糟糕。通常遵循的流程是这样的：

如您所知，问题在于 mydataStoreWriter是一个门面，而真正的写作工作是由 ataskExecutor和 a完成的taskScheduler。通过这些提示到标准输出的消息可以看出这项工作：

如您所见，我为每个 Parquet 文件写入 300K 个对象，这导致磁盘中的文件大约为 700K。没什么大不了的......但是，在一两次写入之后，我收到的消息越来越少，并且进程停止了......

对可能发生的事情有任何想法吗？Cloudera 中的一切都是绿色的……

使用的版本：

• Cloudera 5.11
• 爪哇 8
• Spring 集成 4.3.12.RELEASE
• Spring Data Hadoop 2.2.0.RELEASE

编辑：实际上，我使用 Kite Dataset CLI 工具隔离了 Parquet 文件的编写，问题在于 SKD 本身的性能。使用该csv-import命令并从 CSV 加载数据，我看到我们正在以每分钟 400.000 条记录的速度写入，这远低于我们正在写入的每分钟 15.0000.000 条记录，因此停滞不前......

你能推荐任何提高写作速度的方法吗？谢谢！

我正在尝试使用 Kites SDK 将 JSON 文件转换为 Parquet 格式。

我有以下 pom.xml ：

但是当我尝试构建我的项目时。我收到以下错误：

我无法理解这个问题的根本原因。

我正在尝试将我的 Json 文件转换为 Parquet 格式。

以下是我的 pom 文件。

以下是转换代码：

在上面的代码Movie中是我的 POJO 类。

当我运行程序时，我面临以下异常：

我正在使用 JDK：8。

我没有任何hadoop背景，所以我无法理解它的根本原因。

问题是什么？