1

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.

1 respuesta 1

1

Ya he encontrado la solución a mi problema, para la configuración de Spark agregué Spark Configuration :

return new SparkConf().set("spark.streaming.receiver.writeAheadLog.enable", "true").set("spark.streaming.stopGracefullyOnShutdown" , "true")...;

Parametros para kafka:

auto.offset.reset=latest
enable.auto.commit=false

Para implementar el checkPoint cambié la forma de crear SparkStreaming:

private static JavaStreamingContext createContext(String checkpointDirectory ) {


    System.out.println("Creating new context");
   
    SparkConf sparkConf = new SparkConf().setAppName("JavaRecoverableNetworkWordCount");

    JavaStreamingContext ssc = new JavaStreamingContext(sparkConf, Durations.seconds(5));
    ssc.checkpoint(checkpointDirectory);

    //Codigo con acciones, transformaciones para RDD...

    return ssc;
  }
  
  public static void main(String[] args) throws Exception {

    String checkpointDirectory = "/tmp/directory/checkpoint";
    Function0<JavaStreamingContext> createContextFunc =
        () -> createContext(checkpointDirectory);

    JavaStreamingContext ssc =
      JavaStreamingContext.getOrCreate(checkpointDirectory, createContextFunc);
    ssc.start();
    ssc.awaitTermination();
  }
}

Tu Respuesta

By clicking “Publica tu respuesta”, you agree to our terms of service and acknowledge you have read our privacy policy.

¿No es la respuesta que buscas? Examina otras preguntas con la etiqueta o formula tu propia pregunta.