基于Kafka和Elasticsearch构建实时站内搜索功能的实践

{  "searchable": {    "title": "string",    "content": "string"  },  "metadata": {    "tenant_id": "long",    "type": "long",    "created_at": "date",    "created_by": "string",    "updated_at": "date",    "updated_by": "string"  },  "raw": {}}

Q：为什么不使用Elasticsearch connector之类的连接器对数据进行处理并发送到Elasticsearch中？A：在我们的系统中是不允许将大文本存入到MySQL中的，所以我们使用了额外的对象存储服务来存放我们的产品文档，所以我们无法直接使用连接器将数据发送到Elasticsearch中。Q：为什么不在发送到Kafka前就将数据进行处理？A：这样会有大量的数据被持久化到Kafka中，占用Kafka的磁盘空间，而这部分数据实际上也被存储到了Elasticsearch。Q：为什么要用单独的服务来采集binlog，而不是使用Filebeat之类的agent？A：当然可以直接在MySQL数据库中安装agent来直接采集binlog并发送到Kafka中。但是在部分情况下开发者使用的是云服务商或其他基础设施部门提供的MySQL服务器，这种情况下我们无法直接进入服务器安装agent，所以使用更加通用的、无侵入性的C/S结构来消费MySQL的binlog。

version: "3"services:  mysql:    image: mysql:5.7    container_name: mysql    environment:      MYSQL_ROOT_PASSWORD: root      MYSQL_DATABASE: app    ports:      - 3306:3306    volumes:      - mysql:/var/lib/mysql  zookeeper:    image: bitnami/zookeeper:3.6.2    container_name: zookeeper    ports:      - 2181:2181    volumes:      - zookeeper:/bitnami    environment:      - ALLOW_ANONYMOUS_LOGIN=yes  kafka:    image: bitnami/kafka:2.7.0    container_name: kafka    ports:      - 9092:9092    volumes:      - kafka:/bitnami    environment:      - KAFKA_CFG_ZOOKEEPER_CONNECT=zookeeper:2181      - ALLOW_PLAINTEXT_LISTENER=yes    depends_on:      - zookeeper  elasticsearch:    image: docker.elastic.co/elasticsearch/elasticsearch:7.11.0    container_name: elasticsearch    environment:      - discovery.type=single-node    volumes:      - elasticsearch:/usr/share/elasticsearch/data    ports:      - 9200:9200volumes:  mysql:    driver: local  zookeeper:    driver: local  kafka:    driver: local  elasticsearch:    driver: local

# Elasticsearchcurl "http://localhost:9200/search" -XPUT -d '{  "mappings": {    "properties": {      "searchable": {        "type": "nested",        "properties": {          "title": {            "type": "text"          },          "content": {            "type": "text"          }        }      },      "metadata": {        "type": "nested",        "properties": {          "tenant_id": {            "type": "long"          },          "type": {            "type": "integer"          },          "created_at": {            "type": "date"          },          "created_by": {            "type": "keyword"          },          "updated_at": {            "type": "date"          },          "updated_by": {            "type": "keyword"          }        }      },      "raw": {        "type": "nested"      }    }  }}'

    override fun run() {        client.serverId = properties.serverId        val eventDeserializer = EventDeserializer()        eventDeserializer.setCompatibilityMode(            EventDeserializer.CompatibilityMode.DATE_AND_TIME_AS_LONG        )        client.setEventDeserializer(eventDeserializer)        client.registerEventListener {            val header = it.getHeader<EventHeader>()            val data = it.getData<EventData>()            if (header.eventType == EventType.TABLE_MAP) {                tableRepository.updateTable(Table.of(data as TableMapEventData))            } else if (EventType.isRowMutation(header.eventType)) {                val events = when {                    EventType.isWrite(header.eventType) -> mapper.map(data as WriteRowsEventData)                    EventType.isUpdate(header.eventType) -> mapper.map(data as UpdateRowsEventData)                    EventType.isDelete(header.eventType) -> mapper.map(data as DeleteRowsEventData)                    else -> emptyList()                }                logger.info("Mutation events: {}", events)                for (event in events) {                    kafkaTemplate.send("binlog", objectMapper.writeValueAsString(event))                }            }        }        client.connect()    }

Table: ["id", "title", "content",...]Row: [1, "Foo", "Bar",...]=>{  "id": 1,  "title": "Foo",  "content": "Bar"}

@Componentclass KafkaBinlogTopicListener(    val binlogEventHandler: BinlogEventHandler) {
    companion object {        private val logger = LoggerFactory.getLogger(KafkaBinlogTopicListener::class.java)    }
    private val objectMapper = jacksonObjectMapper()
    @KafkaListener(topics = ["binlog"])    fun process(message: String) {        val binlogEvent = objectMapper.readValue<BinlogEvent>(message)        logger.info("Consume binlog event: {}", binlogEvent)        binlogEventHandler.handle(binlogEvent)    }}

@Componentclass ElasticsearchIndexerBinlogEventHandler(    val restHighLevelClient: RestHighLevelClient) : BinlogEventHandler {    override fun handle(binlogEvent: BinlogEvent) {        val payload = binlogEvent.payload as Map<*, *>        val documentId = "${binlogEvent.database}_${binlogEvent.table}_${payload["id"]}"        // Should delete from Elasticsearch        if (binlogEvent.eventType == EVENT_TYPE_DELETE) {            val deleteRequest = DeleteRequest()            deleteRequest                .index("search")                .id(documentId)            restHighLevelClient.delete(deleteRequest, DEFAULT)        } else {            // Not ever WRITE or UPDATE, just reindex            val indexRequest = IndexRequest()            indexRequest                .index("search")                .id(documentId)                .source(                    mapOf<String, Any>(                        "searchable" to mapOf(                            "title" to payload["title"],                            "content" to payload["content"]                        ),                        "metadata" to mapOf(                            "tenantId" to payload["tenantId"],                            "type" to payload["type"],                            "createdAt" to payload["createdAt"],                            "createdBy" to payload["createdBy"],                            "updatedAt" to payload["updatedAt"],                            "updatedBy" to payload["updatedBy"]                        )                    )                )            restHighLevelClient.index(indexRequest, DEFAULT)        }    }}