-- 創(chuàng)建一個(gè) 用戶名為 maxwell 密碼為 xxxxxx 的用戶
CREATE USER 'maxwell'@'%' IDENTIFIED BY 'XXXXXX';
GRANT ALL ON maxwell.* TO 'maxwell'@'localhost';
GRANT SELECT, REPLICATION CLIENT, REPLICATION SLAVE ON *.* TO 'maxwell'@'%';

# /etc/mysql/my.cnf
[mysqld]
# maxwell 需要的 binlog 格式必須是 row
binlog_format=row
# 指定 server_id 此配置關(guān)系到主從同步需要按情況設(shè)置，
# 由于此mysql沒有開啟主從同步，這邊默認(rèn)設(shè)置為 1
server_id=1
# logbin 輸出的文件名， 按需配置
log-bin=master

sudo systemctl restart mysqld

select @@log_bin;
-- 正確結(jié)果是 1
select @@binlog_format;
-- 正確結(jié)果是 ROW

# /etc/my.cnf
log_slave_updates = 1

-- robin.logs
show create table robin.logs;
-- 表結(jié)構(gòu)
CREATE TABLE `logs` (
  `id` int(11) unsigned NOT NULL AUTO_INCREMENT,
  `content` text NOT NULL,
  `user_id` int(11) NOT NULL,
  `status` enum('SUCCESS','FAILED','PROCESSING') NOT NULL,
  `type` varchar(20) DEFAULT '',
  `meta` text,
  `created_at` bigint(15) NOT NULL,
  `idx_host` varchar(255) DEFAULT '',
  `idx_domain_id` int(11) unsigned DEFAULT NULL,
  `idx_record_value` varchar(255) DEFAULT '',
  `idx_record_opt` enum('DELETE','ENABLED','DISABLED') DEFAULT NULL,
  `idx_orig_record_value` varchar(255) DEFAULT '',
  PRIMARY KEY (`id`),
  KEY `created_at` (`created_at`)
) ENGINE=InnoDB AUTO_INCREMENT=8170697 DEFAULT CHARSET=utf8

wget https://github.com/zendesk/maxwell/releases/download/v1.39.2/maxwell-1.39.2.tar.gz
tar zxvf maxwell-1.39.2.tar.gz **&&**  cd maxwell-1.39.2

./bin/maxwell 
        --host=mysql-maxwell.mysql.svc.cluster.fud3 
        --port=3306 
        --user=root 
        --password=password 
        --replication_host=192.168.5.38 
        --replication_port=3306 
        --replication_user=cloner 
        --replication_password=password
        --filter='exclude: *.*, include: robin.logs' 
        --producer=kafka 
        --kafka.bootstrap.servers=192.168.30.10:9092 
        --kafka_topic=maxwell-robinlogs --kafka_version=0.9.0.1

wget https://artifacts.elastic.co/downloads/logstash/logstash-8.5.0-linux-x86_64.tar.gz
tar zxvf logstash-8.5.0-linux-x86_64.tar.gz

rm config/logstash.yml

# config/logstash-sample.conf
input {
 kafka {
    bootstrap_servers => "192.168.30.10:9092"
    group_id => "main"
    topics => ["maxwell-robinlogs"]
 }
}
filter {
  json {
    source => "message"
  }
  # 將maxwell的事件類型轉(zhuǎn)化為es的事件類型
  # 如增加 -> index 修改-> update
  translate {
    source => "[type]"
    target => "[action]"
    dictionary => {
      "insert" => "index"
      "bootstrap-insert" => "index"
      "update" => "update"
      "delete" => "delete"
    }
    fallback => "unknown"
  }
  # 過濾無效的數(shù)據(jù)
  if ([action] == "unknown") {
    drop {}
  }
  # 處理數(shù)據(jù)格式
  if [data][idx_host] {
    mutate {
      add_field => { "idx_host" => "%{[data][idx_host]}" }
    }
  } else {
    mutate {
      add_field => { "idx_host" => "" }
    }
  }
  if [data][idx_domain_id] {
    mutate {
      add_field => { "idx_domain_id" => "%{[data][idx_domain_id]}" }
    }
  } else {
    mutate {
      add_field => { "idx_domain_id" => "" }
    }
  }
  if [data][idx_record_value] {
    mutate {
      add_field => { "idx_record_value" => "%{[data][idx_record_value]}" }
    }
  } else {
    mutate {
      add_field => { "idx_record_value" => "" }
    }
  }
   if [data][idx_record_opt] {
    mutate {
      add_field => { "idx_record_opt" => "%{[data][idx_record_opt]}" }
    }
  } else {
    mutate {
      add_field => { "idx_record_opt" => "" }
    }
  }
  if [data][idx_orig_record_value] {
    mutate {
      add_field => { "idx_orig_record_value" => "%{[data][idx_orig_record_value]}" }
    }
  } else {
    mutate {
      add_field => { "idx_orig_record_value" => "" }
    }
  }
  if [data][type] {
    mutate {
      replace => { "type" => "%{[data][type]}" }
    }
  } else {
    mutate {
      replace => { "type" => "" }
    }
  }
  mutate {
    add_field => {
      "id" => "%{[data][id]}"
      "content" => "%{[data][content]}"
      "user_id" => "%{[data][user_id]}"
      "status" => "%{[data][status]}"
      "meta" => "%{[data][meta]}"
      "created_at" => "%{[data][created_at]}"
    }
    remove_field => ["data"]
  }
  mutate {
    convert => {
      "id" => "integer"
      "user_id" => "integer"
      "idx_domain_id" => "integer"
      "created_at" => "integer"
    }
  }
  # 只提煉需要的字段
  mutate {
    remove_field => [
      "message",
      "original",
      "@version",
      "@timestamp",
      "event",
      "database",
      "table",
      "ts",
      "xid",
      "commit",
      "tags"
    ]
   }
}
output {
  # 結(jié)果寫到es
  elasticsearch {
    hosts => ["http://es-zico2.service.upyun:9500"]
    index => "robin_logs"
    action => "%{action}"
    document_id => "%{id}"
    document_type => "robin_logs"
  }
  # 結(jié)果打印到標(biāo)準(zhǔn)輸出
  stdout {
    codec => rubydebug
  }
}

# 測試配置文件*
bin/logstash -f config/logstash-sample.conf --config.test_and_exit
# 啟動(dòng)*
bin/logstash -f config/logstash-sample.conf --config.reload.automatic

INSERT INTO maxwell.bootstrap 
        ( database_name, table_name, where_clause, client_id ) 
values 
        ( 'robin', 'logs', 'id > 1', 'maxwell' );

# 檢測 elasticsearch  中的數(shù)據(jù)量
GET robin_logs/robin_logs/_count