匯合HDFS連接器正在丟失消息

社區，請你幫我理解爲什麼〜3％我的消息不會在HDFS？我寫了一個簡單的生產者JAVA來生成1000萬條消息。匯合HDFS連接器正在丟失消息

public static final String TEST_SCHEMA = "{" 
     + "\"type\":\"record\"," 
     + "\"name\":\"myrecord\"," 
     + "\"fields\":[" 
     + " { \"name\":\"str1\", \"type\":\"string\" }," 
     + " { \"name\":\"str2\", \"type\":\"string\" }," 
     + " { \"name\":\"int1\", \"type\":\"int\" }" 
     + "]}"; 

public KafkaProducerWrapper(String topic) throws UnknownHostException { 
    // store topic name 
    this.topic = topic; 

    // initialize kafka producer 
    Properties config = new Properties(); 
    config.put("client.id", InetAddress.getLocalHost().getHostName()); 
    config.put("bootstrap.servers", "myserver-1:9092"); 
    config.put("key.serializer", "io.confluent.kafka.serializers.KafkaAvroSerializer"); 
    config.put("value.serializer", "io.confluent.kafka.serializers.KafkaAvroSerializer"); 
    config.put("schema.registry.url", "http://myserver-1:8089"); 
    config.put("acks", "all"); 

    producer = new KafkaProducer(config); 

    // parse schema 
    Schema.Parser parser = new Schema.Parser(); 
    schema = parser.parse(TEST_SCHEMA); 
} 

public void send() { 
    // generate key 
    int key = (int) (Math.random() * 20); 

    // generate record 
    GenericData.Record r = new GenericData.Record(schema); 
    r.put("str1", "text" + key); 
    r.put("str2", "text2" + key); 
    r.put("int1", key); 

    final ProducerRecord<String, GenericRecord> record = new ProducerRecord<>(topic, "K" + key, (GenericRecord) r); 
    producer.send(record, new Callback() { 
     public void onCompletion(RecordMetadata metadata, Exception e) { 
      if (e != null) { 
       logger.error("Send failed for record {}", record, e); 
       messageErrorCounter++; 
       return; 
      } 
      logger.debug("Send succeeded for record {}", record); 
      messageCounter++; 
     } 
    }); 
} 

public String getStats() { return "Messages sent: " + messageCounter + "/" + messageErrorCounter; } 

public long getMessageCounter() { 
    return messageCounter + messageErrorCounter; 
} 

public void close() { 
    producer.close(); 
} 

public static void main(String[] args) throws InterruptedException, UnknownHostException { 
    // initialize kafka producer 
    KafkaProducerWrapper kafkaProducerWrapper = new KafkaProducerWrapper("my-test-topic"); 

    long max = 10000000L; 
    for (long i = 0; i < max; i++) { 
     kafkaProducerWrapper.send(); 
    } 

    logger.info("producer-demo sent all messages"); 
    while (kafkaProducerWrapper.getMessageCounter() < max) 
    { 
     logger.info(kafkaProducerWrapper.getStats()); 
     Thread.sleep(2000); 
    } 

    logger.info(kafkaProducerWrapper.getStats()); 
    kafkaProducerWrapper.close(); 
}

，而且我用Confluent HDFS Connector在獨立模式將數據寫入到HDFS。的配置如下：

name=hdfs-consumer-test 
connector.class=io.confluent.connect.hdfs.HdfsSinkConnector 
tasks.max=1 

topics=my-test-topic 

hdfs.url=hdfs://my-cluster/kafka-test 
hadoop.conf.dir=/etc/hadoop/conf/ 
flush.size=100000 
rotate.interval.ms=20000 

# increase timeouts to avoid CommitFailedException 
consumer.session.timeout.ms=300000 
consumer.request.timeout.ms=310000 

heartbeat.interval.ms= 60000 
session.timeout.ms= 100000

所述的連接器將數據寫入到HDFS，但等待20000毫秒（由於rotate.interval.ms）之後不是所有的消息被接收。

scala> spark.read.avro("/kafka-test/topics/my-test-topic/partition=*/my-test-topic*") 
    .count() 
res0: Long = 9749015

任何想法是什麼原因造成這種行爲？我的錯誤在哪裏？我正在使用Confluent 3.0.1/Kafka 10.0.0.1。

來源

2016-11-30 Sascha Vetter

你看到最後幾條消息沒有移到HDFS嗎？如果是這樣，很可能您遇到了此處所述的問題https://github.com/confluentinc/kafka-connect-hdfs/pull/100 嘗試在rotate.interval.ms發送後再向該主題發送一條消息過期驗證這是你正在遇到的。如果您需要根據時間進行旋轉，則升級到提取修補程序可能是個好主意。 – dawsaw

這就是解決方案！我更新爲** Confluent 3.1.1 **，我可以在'HDFS'中看到所有消息。你想寫它作爲答案，我給你你應得的榮譽？ –

是的肯定，不知道有真的有區別:) – dawsaw

你看到最後幾條消息沒有移到HDFS嗎？如果是這樣，很可能您遇到此處所述的問題https://github.com/confluentinc/kafka-connect-hdfs/pull/100

嘗試在rotate.interval.ms過期後再向主題發送一條消息以驗證這是您正在運行的內容。如果您需要根據時間進行旋轉，則升級到提取修補程序可能是個好主意。

來源

2016-12-02 00:49:06 dawsaw

匯合HDFS連接器正在丟失消息

回答

相關問題