Vad är asynkron förloppsspårning?

Asynkron förloppsspårning gör att pipelines för strukturerad direktuppspelning kan förlopp för kontrollpunkter asynkront och parallellt med den faktiska databearbetningen i en mikrobatch, vilket minskar svarstiden som är associerad med underhåll av offsetLog och commitLog.

Hur fungerar asynkron förloppsspårning för att minska svarstiden?

Structured Streaming förlitar sig på att bevara och hantera förskjutningar som förloppsindikatorer för frågebearbetning. Förskjutningshanteringsåtgärden påverkar direkt bearbetningens svarstid, eftersom ingen databearbetning kan utföras förrän dessa åtgärder har slutförts. Asynkron förloppsspårning gör att pipelines för strukturerad direktuppspelning kan förlopp för kontrollpunkter utan att påverkas av dessa förskjutningshanteringsåtgärder.

När ska du konfigurera kontrollpunktsfrekvens?

Användare kan konfigurera hur ofta förloppet ska vara kontrollpunkter. Standardinställningarna för kontrollpunktsfrekvens ger bra dataflöde för de flesta frågor. Att konfigurera frekvensen är användbart för scenarier där förskjutningshanteringsåtgärder sker i en högre takt än de kan bearbetas, vilket skapar en ständigt ökande eftersläpning av offsethanteringsåtgärder. För att stoppa den här växande eftersläpningen blockeras eller bromsas databearbetningen, vilket i huvudsak återställer bearbetningsbeteendet för att eliminera fördelarna med asynkron förloppsspårning.

Vilka konfigurationer är associerade med asynkron förloppsspårning?

Hur kan användare aktivera asynkron förloppsspårning?

Användare kan använda kod som liknar koden nedan för att aktivera den här funktionen:

val stream = spark.readStream
      .format("kafka")
      .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
      .option("subscribe", "in")
      .load()

val query = stream.writeStream
     .format("kafka")
        .option("topic", "out")
     .option("checkpointLocation", "/tmp/checkpoint")
        .option("asyncProgressTrackingEnabled", "true")
     .start()

Inaktivera asynkron förloppsspårning

När asynkron förloppsspårning är aktiverat förloppsindikatorer för ramverket inte för varje batch. Innan du inaktiverar asynkron förloppsspårning ska du bearbeta minst två mikrobatchar med följande inställningar:

• .option("asyncProgressTrackingEnabled", "true")
• .option("asyncProgressTrackingCheckpointIntervalMs", 0)

Stoppa frågan när minst två mikrobatchar har slutfört bearbetningen. Nu kan du på ett säkert sätt inaktivera asynkron förloppsspårning och starta om frågan.

Om du har inaktiverat asynkron förloppsspårning utan att slutföra det här steget kan följande fel uppstå:

java.lang.IllegalStateException: batch x doesn't exist

I drivrutinsloggarna kan följande fel visas:

The offset log for batch x doesn't exist, which is required to restart the query from the latest batch x from the offset log. Please ensure there are two subsequent offset logs available for the latest batch via manually deleting the offset file(s). Please also ensure the latest batch for commit log is equal or one batch earlier than the latest batch for offset log.

Genom att följa anvisningarna i det här avsnittet för att inaktivera asynkron förloppsspårning kan du åtgärda dessa fel och reparera din strömningsarbetsbelastning.

Begränsningar med asynkron förloppsspårning

Den här funktionen har följande begränsningar:

• Asynkron förloppsspårning stöds endast i tillståndslösa pipelines när kafka används som mottagare.
• Exakt en gång garanteras inte bearbetning från slutpunkt till slutpunkt med asynkron förloppsspårning eftersom förskjutningsintervall för batch kan ändras i händelse av fel. Vissa handfat, till exempel Kafka, ger aldrig garantier exakt en gång.