Kafkaトピックでレコードが利用できない場合、StreamingQueryStatus.isDataAvailableがオンになるのはなぜですか？

Question

私はspark構造化ストリーミングクエリを停止する必要があります。まずは、カフカで処理できるデータがあるかどうかチェックします。はいの場合は処理してからクエリを停止します。

if (query.status.isDataAvailable) { 
        query.processAllAvailable() 
        } 
query.stop() 

私はJacek Laskowskiの提案を受けて、実装を変更しました。 私は、ストリーミングクエリのリストを持って、クエリのすべての停止後

streamingQueryList.foreach{query => 
query.stop() 
} 

、それは次の例外スロー：目的をデバッグするため

java.lang.InterruptedException 
    at java.util.concurrent.locks.AbstractQueuedSynchronizer.doAcquireSharedInterruptibly(AbstractQueuedSynchronizer.java:998) 
    at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireSharedInterruptibly(AbstractQueuedSynchronizer.java:1304) 
    at scala.concurrent.impl.Promise$DefaultPromise.tryAwait(Promise.scala:202) 
    at scala.concurrent.impl.Promise$DefaultPromise.ready(Promise.scala:218) 
    at scala.concurrent.impl.Promise$DefaultPromise.ready(Promise.scala:153) 
    at org.apache.spark.util.ThreadUtils$.awaitReady(ThreadUtils.scala:222) 
    at org.apache.spark.scheduler.DAGScheduler.runJob(DAGScheduler.scala:621) 
    at org.apache.spark.SparkContext.runJob(SparkContext.scala:2022) 
    at org.apache.spark.sql.execution.datasources.FileFormatWriter$$anonfun$write$1.apply$mcV$sp(FileFormatWriter.scala:179) 
    at org.apache.spark.sql.execution.datasources.FileFormatWriter$$anonfun$write$1.apply(FileFormatWriter.scala:164) 
    at org.apache.spark.sql.execution.datasources.FileFormatWriter$$anonfun$write$1.apply(FileFormatWriter.scala:164) 
    at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:65) 
    at org.apache.spark.sql.execution.datasources.FileFormatWriter$.write(FileFormatWriter.scala:164) 
    at org.apache.spark.sql.execution.streaming.FileStreamSink.addBatch(FileStreamSink.scala:123) 
    at org.apache.spark. 

を、私はから現在のトピックを知るために、クエリのrecentProgressを印刷していますスパークは読書です。そして、私はコンソールコンシューマを使ってそのトピックをチェックしました。そして、kafkaにはデータはありませんが、query.status.isDataAvailableが本当に印刷します。また、たとえ最近のProgressが空であっても、データの処理を待っています。そのため、無限に待っています。

Answer 1

なぜisDataAvailableがtrueになるのか分かりませんが、processAllAvailableはブロックif (query.status.isDataAvailable)で狙っているソリューションを提供します。

processAllAvailableブロックのソースで利用可能なすべてのデータが処理され、シンクにコミットされるまで。

これはあなたの使用例です。ではありませんか？

ただし、"このメソッドはテスト用です。"として"このメソッドは永遠にブロックするかもしれません"（processAllAvailableのscaladocを引用）。

メインスレッドをブロックして（ストリームアプリケーションが起動しているように）、stopクエリ（およびそのバックグラウンドスレッド）を別の「監視」スレッドから都合のよい状態に保つために、代わりにawaitTerminationを使用する必要があります。

awaitTermination（）このクエリの終結待ち、query.stop()によってまたは例外のいずれかによって。クエリが例外で終了した場合、例外がスローされます。