홈>
원래의 경우 답변이 없었기 때문에이 질문을하고 있습니다 : 엘라스틱 카프카 커넥터, ID 생성 .
저도 비슷한 상황입니다.
kafkaconnect를 통해 요청을 보낼 때 단일 필드에 대한 레코드를 작성하지만 여러 필드에 대한 레코드를 작성하는 탄력적 검색 테이블
예외 가져 오기"키는 문서 ID로 사용되며 null 일 수 없습니다"
내 커넥터 구성 :
{
"name": "test-connector33",
"config": {
"connector.class":"io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
"tasks.max": "1",
"topics": "test-connector33",
"connection.url": "http://localhost:9200",
"type.name": "aggregator",
"schema.ignore": "true",
"topic.schema.ignore": "true",
"topic.key.ignore": "false",
"value.converter": "org.apache.kafka.connect.json.JsonConverter",
"value.converter.schemas.enable": "false",
"key.converter": "org.apache.kafka.connect.json.JsonConverter",
"key.converter.schemas.enable": "false",
"key.ignore":"false",
"name": "test-connector33",
"transforms": "InsertKey,extractKey",
"transforms.InsertKey.type":"org.apache.kafka.connect.transforms.ValueToKey",
"transforms.InsertKey.fields":"customerId,city",
"transforms.extractKey.type":"org.apache.kafka.connect.transforms.ExtractField$Key",
"transforms.extractKey.field":"customerId,city"
}}
이 문제를 해결하는 방법이 있습니까?
미리 감사합니다!
- 답변 # 1
관련 자료
- Kafka Connect S3 소스 커넥터를 실행하는 ClassNotFound 예외
- elasticsearch - Python API를 사용하여 탄력적 검색에서 대량 데이터 덤프
- elasticsearch - 문장과 동의어를 비교하는 Elastic Search?
- Kafka Connect 싱크 작업은 허용 한계를 무시합니다
- elasticsearch - 탄력적 검색 고급 클라이언트 - 게시물 요청으로 검색하는 방법?
- amazon web services - 내 VPC/서브넷에서만 Elastic Search 인터넷 엔드 포인트에 액세스
- arrays - Elastic Search for React Native에서 객체를 재구성하는 방법은 무엇입니까?
- azure cosmosdb - Connect Cassandra는 Cosmos Cassandra와 호환됩니까?
- java - Elastic Benstalk에 배포 된 웹 응용 프로그램을 AWS의 MySQL 데이터베이스 인스턴스와 연결하는 방법은 무엇입니까?
- elasticsearch - 탄력적 검색 인기 항목 정렬 및 필터링
- Kafka는 MySQL 사용자 정의 쿼리와 연결
- Elastic Search in ASP.NET - aspnet의 탄력적 검색 - 앰퍼샌드 부호 사용
- Kafka Connect에 두 개의 서로 다른 연결기 배포
- elasticsearch - 탄력적 검색에서 유형 Ip가있는 필드의 와일드 카드 쿼리
- Kafka Connect - 카프카 연결 - 누락 된 텍스트
- elasticsearch - 탄력적 검색 도커 이미지에서 crontab 작업을 실행하는 방법
- elasticsearch - Elastic Search 쿼리 [ImageURL] 쿼리에서 오류가 발생했습니다 쿼리 이름 뒤에 start_object가 없습니다
- 카프카 토픽 생성 명령
- configuration : 원격 서버에서 Elastic Search를 구성하려고 할 때 "master_not_discovered_exception"
- elasticsearch - NER에 탄력적 검색을 사용하고 싶습니다
관련 질문
- Kafka는 "."가 포함된 테이블 이름으로 JDBC 싱크 isse를 연결합니다.
- caching : 연결성이 먼 경우 KAFKA-IGNITE 싱크 커넥터에서 어떤 구성이 필요합니까?
- NoclassDeffoundError는 독립형 모드에서 MongoDB를 위해 Kafka Connect를 실행하는 동안
- Kafka Connect Sink에서 KAFKA 주제의 파티션을 지정하는 방법
- Apache Camel 소스를 사용하여 S3에서 Kafka까지
- 제작자를 통해 메시지로 전달되는 주제에 데이터를 저장하는 방법, 해당 열에 데이터베이스에로드 될 수 있도록
- Kafka는 삽입 필드를 메시지 값으로 변환합니다
- Kafka Connect -TwitterSourceConnector는 SourceConnector의 하위 클래스 여야합니다
- elasticsearch : MySQL Percona 클러스터 용 CDC를 수행하고 모든 DB의 주제를 만드는 방법
- regex : kafka는 정규식을 무시합니다
org.apache.kafka.connect.transforms.ExtractField$Key
단일 필드 만 지원합니다.JSON 객체가
HashMap<String, Object>
인 것처럼 가장하십시오. .customerId,city
필드를 찾을 수 없습니다 그래서map.get(field)
작업은null
를 리턴합니다. 따라서 필드를 null로 설정하십시오.콘솔 제작자를 통해 키를 보내려면
--property print.key=true
를 추가하면됩니다. 플래그로 입력 한 다음 키를 입력하고 tab을 누른 다음 값을 입력하십시오. 프로세스에 데이터를 에코하려면--property key.separator='|'
를 설정할 수도 있습니다 수직 막대의 경우--property parse.key=true
를 추가하십시오.