doris案例的地址
 

在doris中创建表和job

CREATE TABLE logdata.routine_table (
    id      INT            NOT NULL  COMMENT "id",
    name    VARCHAR(30)    NOT NULL  COMMENT "name",
    age     INT                      COMMENT "age"
)
DUPLICATE KEY(`id`)
DISTRIBUTED BY HASH(`id`) BUCKETS 1;

​

CREATE ROUTINE LOAD logdata.routine_job ON routine_table
        PROPERTIES
        (
            "format" = "json"
        )
        FROM KAFKA
        (
            "kafka_broker_list" = "localhost:9092",
            "kafka_topic" = "routineLoad22",
            "property.security.protocol"="SASL_PLAINTEXT",
            "property.sasl.mechanism"="PLAIN",
            "property.sasl.username"="admin",
            "property.sasl.password"="damin"
        );  

--查询当前kafka的所有topic：
/opt/bitnami/kafka/bin/kafka-topics.sh --list --bootstrap-server localhost:9092
​
--创建名为routineLoad22的topic
/opt/bitnami/kafka/bin/kafka-topics.sh --create --bootstrap-server lochost:9092 --topic routineLoad22 --partitions 3 --replication-factor 2
​
--删除名为routineLoad22的topic
/opt/bitnami/kafka/bin/kafka-topics.sh  --delete --bootstrap-server lochost:9092 --topic routineLoad22
​
--导数据
/opt/bitnami/kafka/bin/kafka-console-producer.sh --topic routineLoad22 --bootstrap-server lochost:9092
--查询前十行
/opt/bitnami/kafka/bin/kafka-console-consumer.sh --topic routineLoad22 --bootstrap-server lochost:9092 --from-beginning | head -n 10

以上的命令是建立在kafka的bin目录在/opt/bitnami/kafka/bin这个文件中，具体情况具体分析

输入如下命令：

/opt/bitnami/kafka/bin/kafka-console-producer.sh --topic routineLoad22 --bootstrap-server lochost:9092

导入如下数据

{ "id" : 1, "name" : "Benjamin", "age":18 }
{ "id" : 2, "name" : "Emily", "age":20 }
{ "id" : 3, "name" : "Alexander", "age":22 }

然后在doris中查询

select * from routine_table