kafkacat 및 visidata를 사용하여 Apache Kafka에서 빠른 데이터 프로파일링

다음은 주제에서 100,000개의 JSON 레코드를 샘플링하고 이를 visidata로 파이프합니다.

visidata가 열리면 Shift-F를 눌러 히스토그램을 만듭니다.

kafkacat -b localhost:9092 -t my_topic -C -e -o-100000 | \
  vd --filetype jsonl

visidata가 열리면 화살표 키를 사용하여 히스토그램을 작성하려는 열로 이동하고 Shift-F를 누릅니다. -e 인수에서 kafkacat를 제외하면 파이프와 함께 작동하므로 Kafka 주제에서 메시지의 라이브 스트림을 얻고 visidata는 메시지가 도착할 때 계속 업데이트됩니다(비록 다음과 같은 경우 히스토그램을 다시 그려야 한다고 생각하지만 새로 고침하려는 경우).

데이터가 Avro에 있는 경우 Avro 변환(-s avro) 및 JSON 출력(-J)에 대한 kafkacat의 지원을 대신 사용할 수 있습니다.

kafkacat -b localhost:9092 -t my_topic -C -e -o-100000 \
  -r http://schema-registry:8081 -s avro -J | \
  jq -c '.payload'| \
  vd --filetype jsonl

필드는 중첩될 수 있습니다. 확장하려면 visidata에서 g(를 사용하십시오.

Confluent Cloud와 함께 사용

원시 JSON 메시지:

kafkacat -X security.protocol=SASL_SSL -X sasl.mechanisms=PLAIN \
         -b BROKER.gcp.confluent.cloud:9092 \
         -X sasl.username="CCLOUD_API_KEY" \
         -X sasl.password="CCLOUD_API_PASSWORD" \
         -t my_topic -C -e -o-10000 | \
         vd --filetype jsonl

Avro 데이터(Confluent Cloud의 스키마 레지스트리):

kafkacat -X security.protocol=SASL_SSL -X sasl.mechanisms=PLAIN \
         -b BROKER.gcp.confluent.cloud:9092 \
         -X sasl.username="CCLOUD_API_KEY" \
         -X sasl.password="CCLOUD_API_PASSWORD" \
         -s avro \
         -r https://SR_API_KEY:SR_API_SECRET@SR_ENDPOINT.gcp.confluent.cloud \
         -t my_avro_topic -C -e -o-10000 | \
         vd --filetype jsonl

참고: 스키마 레지스트리 URL에 자격 증명을 제공할 때 자격 증명을 URL 인코딩해야 합니다(a6kme for this tip! 덕분에). 그렇지 않으면 오류Avro/Schema-registry message deserialization: REST request failed (code -1): HTTP request failed: URL using bad/illegal format or missing URL : terminating가 발생할 수 있습니다.

따라서 Schema Registry API 키와 비밀이 key123! 및 S3cr3t/kjna%$!%dsf£인 경우 URL encode it 및 https://key123%21:S3cr3t%2Fkjna%25%24%21%25dsf%C2%A3@SR_ENDPOINT.gcp.confluent.cloud를 사용합니다.