Я хочу послушать кафка-топик. Каждое событие по этой теме представляет, что запись моего ведра S3 может быть удалена.private static Dataset readDataFromKafka(SparkSession sparkSession)
{
return sparkSession.readStream()
.format("kafka")
.option("kafka.bootstrap.servers",
"")
.option("subscribe", "")
.option("", "")
.load()
;
}
< /code>
Чтение событий, кажется, функционирует, как я представлял. Предположим, это просто идентификатор. < /P>
Dataset deltatable = readDataFromKafka(sparkSession);
Dataset idDataset = deltatable.as(Encoders.bean(ID.class));
< /code>
Следующий фрагмент кода предназначен для удаления событий из моего Deltatable: < /p>
StreamingQuery deleteQuery = idDataset.writeStream()
.foreach(new DeleteForeachWriter(S3BUCKET))
.outputMode("append")
.option("checkpointLocation", S3BUCKET + "/checkpoints/delete")
.start();
deleteQuery.awaitTermination();
< /code>
my deleteforeachwriter выглядит так: < /p>
public class DeleteForeachWriter extends ForeachWriter {
private DeltaTable deltaTable;
public DeleteForeachWriter(String deltaTablePath) {
this.deltaTable = DeltaTable.forPath(spark, deltaTablePath);
}
@Override
public boolean open(long partitionId, long version) {
// Open connection (if needed)
return true;
}
@Override
public void process(ID event) {
// Perform the delete operation
deltaTable.delete("id = '" + "ID.getId()" + "'");
}
@Override
public void close(Throwable errorOrNull) {
// Close connection (if needed)
}
}
< /code>
После запуска применения я получаю следующее исключение: < /p>
error utils: Aborting task
орг.apache.spark.sql.delta.deltaillegalstatexcept org.apache.spark.sql.delta.deltaerrorsbase.deltatablefoundinexecutor (deltaerrors.scala: 2743)
at org.apache.spark.sql.delta.deltaerrorsbase.deltatablefoundexecutor $ (deltaerrors.scala: 272). org.apache.spark.sql.delta.deltaerrors $ .deltatablefoundinexecutor (deltrorrors.scala: 3598)
at io.delta.tables.deltatable.df (deltatable.scala: 61)
at. io.delta.tables.deltatable.todf (deltatable.scala: 88)
at io.delta.tables.excution.deltatableoperations.sparksession (deltatableoperations.scala: 169)
at io.delta.tables.execution.deltatableoperations.sparksession $ (deltatableoperations.scala: 169)
at io.delta.tables.deltatable.sparksession (deltatable.scala: 44)
at io.delta.tables.execution.deltatableoperations. $ anonfun $ cerfeledete 1 (deltatableoperations.scala: 44)
at org.apache.spark.sql.delta.util.analysishelper.improveunsupportedoperror (alyrysishelper.scala: 109)
at.scal org.apache.spark.sql.delta.util.analysishelper.improveunsupportedoPerror $ (AnalySishelper.scala: 95)
at io.delta.tables.deltatatable.improveunsupporteperror (deltatable.scala: 44)
io.delta.tables.execution.deltatableOperations.executedElete (deltatableOporations.scala: 44)
at io.delta.tables.excution.deltatableoperations.executedelete $ io.delta.tables.deltatable.executedelete (deltatable.scala: 44)
at io.delta.tables.deltatable.delete (deltatable.scala: 184)
at io.delta.tables.deltatable.delete (deltatable.scala: 173)
>
Подробнее здесь: https://stackoverflow.com/questions/796 ... ed-on-attr
