kafka의 topic 다분구 상황, 어떻게 크로스 정보 소비의 순서성을 확보합니까

이 문제는 엄밀히 말하면 틀림없이 있을 것이다.kafka는 단지 구역 내의 질서성을 보장할 수 있을 뿐이다.
 
다음은 카프카 저자 제이 크렙스의 블로그에서 카프카의 디자인 마인드를 소개하는 대목이다.
Each partition is a totally ordered log, but there is no global ordering between partitions (other than perhaps some wall-clock time you might include in your messages). The assignment of the messages to a particular partition is controllable by the writer, with most users choosing to partition by some kind of key (e.g. user id). Partitioning allows log appends to occur without co-ordination between shards and allows the throughput of the system to scale linearly with the Kafka cluster size.
 
일부 메시지의 질서정연함(message.key와 같은 메시지는 소비 순서를 확보해야 함) 장면에 대해 제품이kafka에 데이터를 삽입할 때 제어하고 같은 키가 같은partition에 나누어진다.
 
kafka 원본 코드는 다음과 같습니다. 이 방식을 지원합니다.

private[kafka]classDefaultPartitioner[T]extendsPartitioner[T]{
  privateval random = newjava.util.Random
  def partition(key: T, numPartitions: Int): Int = {
    if(key== null){
        println("key is null")
        random.nextInt(numPartitions)
    }
    else{
        println("key is "+ key + " hashcode is "+key.hashCode)
        math.abs(key.hashCode) % numPartitions
    }
  }
}

 
 
kafka-storm에서 원 파티션 -> 원 consumer instance라면 이런 문제가 없지만 병행을 잃었습니다.
N1 partitions -> N2 consumer instances 의 경우
1)N1
2) N1>N2(N2>1), 이런 경우 각kafka-spout의 실례는 고정된 1개 또는 몇 개의partition을 소비하고 msg는 서로 다른consumer에 의해 중복 소비되지 않는다.
3) N1=N2, 이 경우 실제 작업에서 1개의consumer instance가 1개의partition을 소비해야 한다는 것을 발견했다.1개의partition은 1개의consumer 실례만 있을 수 있으며, 그렇지 않으면 자물쇠를 잠그는 등 조작이 필요하기 때문에 소비 통제의 복잡성을 줄일 수 있다.
 
 
장면 적용:
특정 위치에서 사용자의 체류 시간을 계산하고 로그 내용을 사용자 ID, 시점, 위치로 추상화할 수 있습니다.
애플리케이션 - 로그 파일 sftp 서버 - 데이터 수집 계층 - kafka - storm 실시간 데이터 세척 처리 계층 - Redis, Hbase - 시간제 작업, mapreduce
집적 테스트 기간에 실제 로그가 없기 때문에 채집층 시뮬레이션에서kafka에 데이터를 삽입(특히 송신 주파수 시뮬레이션은 매우 거칠다)한 결과 실시간 처리층에서 사용자가 어느 위치에서 체류하는 시간을 마이너스로 계산한 것을 발견했다. 원인은 다음과 같다.
 
1) 채집층의 시뮬레이션은 진실하지 않다(같은 사용자가 카프카에 삽입한 위치의 시간은 무작위로 생성된다). 그러나 현재의 로그 파일 sftp 서버나 채집층에 이런 상황이 있는지를 고려하여 업무 차원에서 회피하고 이 무효 데이터를 필터할 수 있다.
 
2)storm에서tuple 처리 실패, 재발행,kafka-storm에서offset을 실패한 위치로 되돌려보냈지만 이전의 위치 정보는 리디스에 캐시되었을 수 있다(hbase 방문 횟수를 줄이기 위해 사용자의 최근 위치 정보를 리디스에 두었다). 그러면 오프셋 이후의 모든 정보는 다시 소비되고 이로써 체류 시간은 마이너스이다.레디스에 저장하지 않고 이 기록을 필터할 수 있습니다. 
 
실제 데이터: U1 T1 A1->U1 T2 A2
 
fail 재발급: U1 T1 A1->U1 T2 A2 -> 앞 두 개 모두 실패, 재발급-> U1 T1 A1 (마이너스 체류 시간)-> U1 T2 A2
 
실패한 재발견이기 때문에 at least once입니다. only once라면 이런 상황이 없습니다.
 
 
PS: 일부 원리적인 문제는 "kafka 소비 원리"를 참고하여 소개할 수 있습니다.