Buen día, Me encuentro aprendiendo Spark y Kafka, tengo una aplicación streaming usando el API para java de spark. Mi problema radica en ¿cómo poder conocer cual fue el último offset leído? Por ejemplo mi aplicación lee constantemente de kafka, de pronto se cae y en el topic kafka llegan mensajes. Quiero hacer que cuando se levante mi aplicación Spark, continúe leyendo apartir del offset que se leyó antes de la caída. Con kafkaUtils si coloco:
auto.offset.reset=earliest
Me lee todo el topic, si lo cambio a latest me lee los records que llegan una vez se inicia la aplicación, por lo que no cumple con lo que quiero, ya que tendría perdida de datos de kafka. Agregue un checkPoint, entiendo que esto funciona cuando no se termino de leer un DStream, si se cae puede continuar con un trabajo que se vio interrumpido.
Código de configuración:
Spark config:
return new SparkConf().setMaster(sparkDriverUtils.getSparkMaster())
.setAppName(sparkDriverUtils.getSparkAppName() == null ? "name-not-set"
: sparkDriverUtils.getSparkAppName())
.set("spark.default.parallelism", "2")
.set("spark.yarn.maxAppAttempts", "1")
.set("spark.yarn.am.attemptFailuresValidityInterval", "2h")
.set(DriverConstants.SPARK_MAX_CORES, sparkDriverUtils.getSparkMaxCores())
.set(DriverConstants.SPARK_EXECUTOR_MEMORY, sparkDriverUtils.getSparkExecutorMemory());
Fragmento para Kafka: Parametros Kafka :
kafkaParams.put(DriverConstants.KAFKA_BOOTSTRAP_SERVER, getBootstrapServers());// 1 or more brokers
kafkaParams.put(DriverConstants.KEY_DESERIALIZER, StringDeserializer.class);
kafkaParams.put(DriverConstants.VALUE_DESERIALIZER, StringDeserializer.class);
kafkaParams.put(DriverConstants.KAFKA_GROUP_ID, getGroupId());
kafkaParams.put(DriverConstants.KAFKA_AUTO_OFFSET_RESET, getAutoOffsetReset()); //earliest o latest
kafkaParams.put(DriverConstants.KAFKA_ENABLE_AUTO_COMMIT, getEnableAutoCommit()); //false
JavaStreamingContext jssc = sparkConfigurationBuilder
.buildJSC(sparkConfigurationBuilder.buildSparkConfiguration());
jssc.checkpoint("C:\\tmp\\poc\\checkPoint");
Map<String, Object> kafkaParams = sparkDriverUtils.getKafkaProperties();
Collection<String> topics = Arrays.asList(sparkDriverUtils.getTopics().trim().split(","));// 1 o more topics
JavaInputDStream<ConsumerRecord<String, String>> stream = KafkaUtils.createDirectStream(jssc,
LocationStrategies.PreferConsistent(),
ConsumerStrategies.<String, String>Subscribe(topics, kafkaParams));
Saludos.