問題タブ [kite-sdk]

HDFS から何百万ものレコードを読み取り、それらを処理し、AVRO 形式で HDFS に書き戻す Spark ジョブがあります。多くのファイル (書き込まれた) が .avro.tmp 状態のままであることが観察されました。

AVRO 形式でデータを書き込むために Kite SDK を使用しています。環境はCDH5.5です。

レコードの読み取りとエグゼキュータへの送信が完了するとすぐに、Spark ジョブが終了するためである可能性があります (実際に書き込みを行うのはどちらですか?)。

その場合、すべての .tmp が .avro に変換されるまでジョブが終了しないようにするにはどうすればよいですか? それとも他に何が原因でしょうか？

私は NIFI の初心者です。私の使用例は、ポートから読み取り、寄木細工形式で hdfs に書き込むことです。私の調査によると、KiteSDK と呼ばれるものがあり、これを使用して Parquet 形式で保存できます。よろしいですか?アドバイスをお願いします。どんな例でも役に立ちます。

Ubuntu-14.04でHDP-2.5を使用して、このコマンドを実行し、

import raw csvKiteSdk を使用して Hive にデータを送信しようとするver.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: ファイル file:/hdp/apps/2.5.0.0-1245/mapreduce/mapreduce.tar.gz は org.apache.hadoop.fs.RawLocalFileSystem.deprecatedGetFileStatus(RawLocalFileSystem.java:624) at org に存在しません.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 org.apache.org.apache.crunch.hadoop.mapreduce.lib.jobcontrol.CrunchControlledJob の org.apache.hadoop.mapreduce.Job.submit(Job.java:1287) の hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1724)。 org.apache.crunch.hadoop.mapreduce.lib.jobcontrol.CrunchJobControl の org.apache.crunch.hadoop.mapreduce.lib.jobcontrol.CrunchJobControl.startReadyJobs(CrunchJobControl.java:204) で (CrunchControlledJob.java:329) を送信します。 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) で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)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)org.apache.crunch.hadoop.mapreduce.lib.jobcontrol.CrunchControlledJob.submit(Cru​​nchControlledJob.java:329) の org.apache.crunch.hadoop.mapreduce.lib の mapreduce.Job.submit(Job.java:1287)。 org.apache.crunch.impl.mr.exec の org.apache.crunch.hadoop.mapreduce.lib.jobcontrol.CrunchJobControl.pollJobStatusAndStartNewOnes(CrunchJobControl.java:238) で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. java.lang.Thread.run(Thread.java:745) で run(MRExecutor.java:83)org.apache.crunch.hadoop.mapreduce.lib.jobcontrol.CrunchControlledJob.submit(Cru​​nchControlledJob.java:329) の org.apache.crunch.hadoop.mapreduce.lib の mapreduce.Job.submit(Job.java:1287)。 org.apache.crunch.impl.mr.exec の org.apache.crunch.hadoop.mapreduce.lib.jobcontrol.CrunchJobControl.pollJobStatusAndStartNewOnes(CrunchJobControl.java:238) で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. java.lang.Thread.run(Thread.java:745) で run(MRExecutor.java:83)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) で java.lang.Thread.run(Thread.java:745) で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) で java.lang.Thread.run(Thread.java:745) で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)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)

ファイル"hdfs:/hdp/apps/2.5.0.0-1245/mapreduce/mapreduce.tar.gz" が存在することを確認しましたが、このエラーを解決する方法がしばらくわかりません。

どんな助けでも大歓迎です。

Flume (1.6 & 1.7) の作業中に、以下のエラーが発生します。

私のエージェント パイプラインは、spooldir ソースから json データを取り込み、モルライン インターセプターを使用して抽出および変換するように設定されています。

以下のflume構成の抜粋を参照してください

私は NiFi でパイプラインを設定しています。ここで JSON レコードを取得し、それを使用して API にリクエストを送信します。私が受け取る応答には、数値データとテキスト データの両方が含まれます。次に、このデータを Hive に書き込む必要があります。私InferAvroSchemaはスキーマを推測するために使用します。一部の数値は次のような符号付きの値です。プロセッサは、型を推測している間、それらをor またはor型ではない-2.46,-0.1と見なします。stringdoublefloatdecimal

プロセッサで AVRO スキーマをハードコーディングできることはわかっていますが、 を利用してより動的にする方がさらに良いと思いInferAvroSchemaました。これを克服/解決できる他の方法はありますか?