基于 Kafka 技术栈构建和部署实时搜索引擎的实践

CREATE STREAM "brands"WITH (    kafka_topic = 'store.public.brands',     value_format = 'avro');

CREATE STREAM "enriched_brands"WITH (    kafka_topic = 'enriched_brands') AS     SELECT         CAST(brand.id AS VARCHAR) as "id",         brand.tenant_id as "tenant_id",        brand.name as "name"     FROM         "brands" brand     PARTITION BY         CAST(brand.id AS VARCHAR)     EMIT CHANGES;

CREATE TABLE "brands_table"AS     SELECT         id as "id",         latest_by_offset(tenant_id) as "tenant_id"    FROM         "brands" group by id     EMIT CHANGES;    

CREATE STREAM "brand_products" WITH (    kafka_topic = 'store.public.brand_products',     value_format = 'avro' );

CREATE STREAM "enriched_brand_products" WITH (    kafka_topic = 'enriched_brand_products’ ) AS     SELECT         "brand"."id" as "brand_id",         "brand"."tenant_id" as "tenant_id",         CAST(brand_product.id AS VARCHAR) as "id",        brand_product.name AS "name"    FROM         "brand_products" AS brand_product     INNER JOIN "brands_table" "brand"    ON         brand_product.brand_id = "brand"."id"    PARTITION BY         CAST(brand_product.id AS VARCHAR)     EMIT CHANGES;

postgres:    build: services/postgres    container_name: oeso_postgres    volumes:      - database:/var/lib/postgresql/data    env_file:      - .env    ports:      - 5432:5432    networks:      - project_network      

elasticsearch:    image: docker.elastic.co/elasticsearch/elasticsearch:7.10.0    container_name: elasticsearch    volumes:      - ./services/elasticsearch/config/elasticsearch.yml:/usr/share/elasticsearch/config/elasticsearch.yml:ro      - elasticsearch-database:/usr/share/elasticsearch/data    env_file:      - .env    ports:      - "9200:9200"      - "9300:9300"    networks:      - project_network

zookeeper:    image: confluentinc/cp-zookeeper:6.0.0    hostname: zookeeper    container_name: zookeeper    ports:      - "2181:2181"    environment:      ZOOKEEPER_CLIENT_PORT: 2181      ZOOKEEPER_TICK_TIME: 2000    networks:      - project_network

broker:    image: confluentinc/cp-enterprise-kafka:6.0.0    hostname: broker    container_name: broker    depends_on:      - zookeeper    ports:      - "29092:29092"    environment:      KAFKA_BROKER_ID: 1      KAFKA_ZOOKEEPER_CONNECT: "zookeeper:2181"      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://broker:9092,PLAINTEXT_HOST://localhost:29092      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1    networks:      - project_network

schema-registry:    image: confluentinc/cp-schema-registry:6.0.0    hostname: schema-registry    container_name: schema-registry    depends_on:      - zookeeper      - broker    ports:      - "8081:8081"    environment:      SCHEMA_REGISTRY_HOST_NAME: schema-registry      SCHEMA_REGISTRY_KAFKASTORE_CONNECTION_URL: "zookeeper:2181"    networks:      - project_network

connect:    image: confluentinc/cp-kafka-connect:6.0.0    hostname: connect    container_name: connect    volumes:      - "./producers/debezium-debezium-connector-postgresql/:/usr/share/confluent-hub-components/debezium-debezium-connector-postgresql/"      - "./consumers/confluentinc-kafka-connect-elasticsearch/:/usr/share/confluent-hub-components/confluentinc-kafka-connect-elasticsearch/"    depends_on:      - zookeeper      - broker      - schema-registry    ports:      - "8083:8083"    environment:      CONNECT_BOOTSTRAP_SERVERS: "broker:9092"      KAFKA_HEAP_OPTS: "-Xms256M -Xmx512M"      CONNECT_REST_ADVERTISED_HOST_NAME: connect      CONNECT_REST_PORT: 8083      CONNECT_GROUP_ID: compose-connect-group      CONNECT_CONFIG_STORAGE_TOPIC: docker-connect-configs      CONNECT_CONFIG_STORAGE_REPLICATION_FACTOR: 1      CONNECT_OFFSET_FLUSH_INTERVAL_MS: 10000      CONNECT_OFFSET_STORAGE_TOPIC: docker-connect-offsets      CONNECT_OFFSET_STORAGE_REPLICATION_FACTOR: 1      CONNECT_STATUS_STORAGE_TOPIC: docker-connect-status      CONNECT_STATUS_STORAGE_REPLICATION_FACTOR: 1      CONNECT_KEY_CONVERTER: org.apache.kafka.connect.storage.StringConverter      CONNECT_VALUE_CONVERTER: io.confluent.connect.avro.AvroConverter      CONNECT_VALUE_CONVERTER_SCHEMA_REGISTRY_URL: http://schema-registry:8081      CONNECT_INTERNAL_KEY_CONVERTER: "org.apache.kafka.connect.json.JsonConverter"      CONNECT_INTERNAL_VALUE_CONVERTER: "org.apache.kafka.connect.json.JsonConverter"      CONNECT_ZOOKEEPER_CONNECT: "zookeeper:2181"      CLASSPATH: /usr/share/java/monitoring-interceptors/monitoring-interceptors-5.5.1.jar      CONNECT_PRODUCER_INTERCEPTOR_CLASSES: "io.confluent.monitoring.clients.interceptor.MonitoringProducerInterceptor"      CONNECT_CONSUMER_INTERCEPTOR_CLASSES: "io.confluent.monitoring.clients.interceptor.MonitoringConsumerInterceptor"      CONNECT_PLUGIN_PATH: "/usr/share/java,/usr/share/confluent-hub-components"      CONNECT_LOG4J_LOGGERS: org.apache.zookeeper=ERROR,org.I0Itec.zkclient=ERROR,org.reflections=ERROR    networks:      - project_network

ksqldb-server:    image: confluentinc/cp-ksqldb-server:6.0.0    hostname: ksqldb-server    container_name: ksqldb-server    depends_on:      - broker      - schema-registry    ports:      - "8088:8088"    volumes:      - "./producers/debezium-debezium-connector-postgresql/:/usr/share/kafka/plugins/debezium-debezium-connector-postgresql/"      - "./consumers/confluentinc-kafka-connect-elasticsearch/:/usr/share/kafka/plugins/confluentinc-kafka-connect-elasticsearch/"    environment:      KSQL_LISTENERS: "http://0.0.0.0:8088"      KSQL_BOOTSTRAP_SERVERS: "broker:9092"      KSQL_KSQL_SCHEMA_REGISTRY_URL: "http://schema-registry:8081"      KSQL_KSQL_LOGGING_PROCESSING_STREAM_AUTO_CREATE: "true"      KSQL_KSQL_LOGGING_PROCESSING_TOPIC_AUTO_CREATE: "true"      KSQL_KSQL_STREAMS_MAX_TASK_IDLE_MS: 2000      KSQL_CONNECT_GROUP_ID: "ksql-connect-cluster"      KSQL_CONNECT_BOOTSTRAP_SERVERS: "broker:9092"      KSQL_CONNECT_KEY_CONVERTER: "io.confluent.connect.avro.AvroConverter"      KSQL_CONNECT_VALUE_CONVERTER: "io.confluent.connect.avro.AvroConverter"      KSQL_CONNECT_KEY_CONVERTER_SCHEMA_REGISTRY_URL: "http://schema-registry:8081"      KSQL_CONNECT_VALUE_CONVERTER_SCHEMA_REGISTRY_URL: "http://schema-registry:8081"      KSQL_CONNECT_VALUE_CONVERTER_SCHEMAS_ENABLE: "false"      KSQL_CONNECT_CONFIG_STORAGE_TOPIC: "ksql-connect-configs"      KSQL_CONNECT_OFFSET_STORAGE_TOPIC: "ksql-connect-offsets"      KSQL_CONNECT_STATUS_STORAGE_TOPIC: "ksql-connect-statuses"      KSQL_CONNECT_CONFIG_STORAGE_REPLICATION_FACTOR: 1      KSQL_CONNECT_OFFSET_STORAGE_REPLICATION_FACTOR: 1      KSQL_CONNECT_STATUS_STORAGE_REPLICATION_FACTOR: 1      KSQL_CONNECT_PLUGIN_PATH: "/usr/share/kafka/plugins"    networks:      - project_network

ksqldb-cli:    image: confluentinc/cp-ksqldb-cli:6.0.0    container_name: ksqldb-cli    depends_on:      - broker      - ksqldb-server    entrypoint: /bin/sh    tty: true    networks:      - project_network

streams-init:    build: jobs/streams-init    container_name: streams-init    depends_on:      - zookeeper      - broker      - schema-registry      - ksqldb-server      - ksqldb-cli      - postgres      - elasticsearch      - connect    env_file:      - .env    environment:      ZOOKEEPER_HOSTS: "zookeeper:2181"      KAFKA_TOPICS: "brands, brand_products"    networks:      - project_network

#!/bin/bash

# Setup ENV variables in connectors json filessed -i "s/POSTGRES_USER/${POSTGRES_USER}/g" connectors/postgres.jsonsed -i "s/POSTGRES_PASSWORD/${POSTGRES_PASSWORD}/g" connectors/postgres.jsonsed -i "s/POSTGRES_DB/${POSTGRES_DB}/g" connectors/postgres.jsonsed -i "s/ELASTIC_PASSWORD/${ELASTIC_PASSWORD}/g" connectors/elasticsearch.json

# Simply wait until original kafka container and zookeeper are started.export WAIT_HOSTS=zookeeper:2181,broker:9092,schema-registry:8081,ksqldb-server:8088,elasticsearch:9200,connect:8083export WAIT_HOSTS_TIMEOUT=300/wait

# Parse string of kafka topics into an array# https://stackoverflow.com/a/10586169/4587961kafkatopicsArrayString="$KAFKA_TOPICS"IFS=', ' read -r -a kafkaTopicsArray <<< "$kafkatopicsArrayString"

# A separate variable for zookeeper hosts.zookeeperHostsValue=$ZOOKEEPER_HOSTS

# Terminate all queriescurl -s -X "POST" "http://ksqldb-server:8088/ksql" \         -H "Content-Type: application/vnd.ksql.v1+json; charset=utf-8" \         -d '{"ksql": "SHOW QUERIES;"}' | \  jq '.[].queries[].id' | \  xargs -Ifoo curl -X "POST" "http://ksqldb-server:8088/ksql" \           -H "Content-Type: application/vnd.ksql.v1+json; charset=utf-8" \           -d '{"ksql": "TERMINATE 'foo';"}'           

# Drop All Tablescurl -s -X "POST" "http://ksqldb-server:8088/ksql" \             -H "Content-Type: application/vnd.ksql.v1+json; charset=utf-8" \             -d '{"ksql": "SHOW TABLES;"}' | \      jq '.[].tables[].name' | \      xargs -Ifoo curl -X "POST" "http://ksqldb-server:8088/ksql" \               -H "Content-Type: application/vnd.ksql.v1+json; charset=utf-8" \               -d '{"ksql": "DROP TABLE \"foo\";"}'



# Drop All Streamscurl -s -X "POST" "http://ksqldb-server:8088/ksql" \           -H "Content-Type: application/vnd.ksql.v1+json; charset=utf-8" \           -d '{"ksql": "SHOW STREAMS;"}' | \    jq '.[].streams[].name' | \    xargs -Ifoo curl -X "POST" "http://ksqldb-server:8088/ksql" \             -H "Content-Type: application/vnd.ksql.v1+json; charset=utf-8" \             -d '{"ksql": "DROP STREAM \"foo\";"}'             

# Create kafka topic for each topic item from split array of topics.for newTopic in "${kafkaTopicsArray[@]}"; do    # https://kafka.apache.org/quickstart    curl -X DELETE http://elasticsearch:9200/enriched_$newTopic --user elastic:${ELASTIC_PASSWORD}    curl -X DELETE http://schema-registry:8081/subjects/store.public.$newTopic-value    kafka-topics --create --topic "store.public.$newTopic" --partitions 1 --replication-factor 1 --if-not-exists --zookeeper "$zookeeperHostsValue"    curl -X POST -H "Content-Type: application/vnd.schemaregistry.v1+json" --data @schemas/$newTopic.json http://schema-registry:8081/subjects/store.public.$newTopic-value/versions

done

curl -X "POST" "http://ksqldb-server:8088/ksql" -H "Accept: application/vnd.ksql.v1+json" -d 
{ "ksql": "CREATE STREAM \\"brands\\" WITH (kafka_topic = \'store.public.brands\', value_format = \'avro\');", "streamsProperties": {} }'curl -X "POST" "http://ksqldb-server:8088/ksql" -H "Accept: application/vnd.ksql.v1+json" -d 
{ "ksql": "CREATE STREAM \\"enriched_brands\\" WITH ( kafka_topic = \'enriched_brands\' ) AS SELECT CAST(brand.id AS VARCHAR) as \\"id\\", brand.tenant_id as \\"tenant_id\\", brand.name as \\"name\\" from \\"brands\\" brand partition by CAST(brand.id AS VARCHAR) EMIT CHANGES;", "streamsProperties": {} }'

curl -X "POST" "http://ksqldb-server:8088/ksql" -H "Accept: application/vnd.ksql.v1+json" -d 
{ "ksql": "CREATE STREAM \\"brand_products\\" WITH ( kafka_topic = \'store.public.brand_products\', value_format = \'avro\' );", "streamsProperties": {} }'curl -X "POST" "http://ksqldb-server:8088/ksql" -H "Accept: application/vnd.ksql.v1+json" -d 
{ "ksql": "CREATE TABLE \\"brands_table\\" AS SELECT id as \\"id\\", latest_by_offset(tenant_id) as \\"tenant_id\\" FROM \\"brands\\" group by id EMIT CHANGES;", "streamsProperties": {} }'curl -X "POST" "http://ksqldb-server:8088/ksql" -H "Accept: application/vnd.ksql.v1+json" -d 
{ "ksql": "CREATE STREAM \\"enriched_brand_products\\" WITH ( kafka_topic = \'enriched_brand_products\' ) AS SELECT \\"brand\\".\\"id\\" as \\"brand_id\\", \\"brand\\".\\"tenant_id\\" as \\"tenant_id\\", CAST(brand_product.id AS VARCHAR) as \\"id\\", brand_product.name AS \\"name\\" FROM \\"brand_products\\" AS brand_product INNER JOIN \\"brands_table\\" \\"brand\\" ON brand_product.brand_id = \\"brand\\".\\"id\\" partition by CAST(brand_product.id AS VARCHAR) EMIT CHANGES;", "streamsProperties": {} }'

curl -X DELETE http://connect:8083/connectors/enriched_writercurl -X "POST" -H "Content-Type: application/json" --data @connectors/elasticsearch.json http://connect:8083/connectors

curl -X DELETE http://connect:8083/connectors/event_readercurl -X "POST" -H "Content-Type: application/json" --data @connectors/postgres.json http://connect:80

cp default.env .envdocker-compose up -d

CREATE TABLE brands (    id serial PRIMARY KEY,    name VARCHAR (50),    tenant_id INTEGER);CREATE TABLE brand_products (    id serial PRIMARY KEY,    brand_id INTEGER,    name VARCHAR(50));

INSERT INTO brands VALUES(1, 'Brand Name 1', 1);INSERT INTO brands VALUES(2, 'Brand Name 2', 1);INSERT INTO brands VALUES(3, 'Brand Name 3', 2);INSERT INTO brands VALUES(4, 'Brand Name 4', 2);

INSERT INTO brand_products VALUES(1, 1, 'Product Name 1');INSERT INTO brand_products VALUES(2, 2, 'Product Name 2');INSERT INTO brand_products VALUES(3, 3, 'Product Name 3');INSERT INTO brand_products VALUES(4, 4, 'Product Name 4');INSERT INTO brand_products VALUES(5, 1, 'Product Name 5');

curl localhost:9200/enriched_brand_products/_search --user elastic:your_password