文章目錄

資料同步ElasticSearch
- 單表基本配置
- 配接器映射檔案詳細介紹（單表、多表映射介紹）
- - 單表映射索引示例sql
  - 單表映射索引示例sql帶函式或運算操作
  - 多表映射(一對一, 多對一)索引示例sql
  - 多表映射(一對多)索引示例sql
  - 其它型別的sql示例
  - 注意事項

本文詳細介紹Canal 配置保存 ElasticSearch

Canal從零配置使用參考：https://blog.csdn.net/zhangshenghang/article/details/120361721

資料同步ElasticSearch

我們接著在之前配置Hbase基礎上直接修改配置，實作同時同步ElasticSearch

單表基本配置

1.修改啟動器配置 {canal-apapter}/conf/application.yml

server:
  port: 8081
logging:
  level:
    com.alibaba.otter.canal.client.adapter: DEBUG
    com.alibaba.otter.canal.client.adapter.hbase: DEBUG
spring:
  jackson:
    date-format: yyyy-MM-dd HH:mm:ss
    time-zone: GMT+8
    default-property-inclusion: non_null
canal.conf:
  # tcp kafka rocketMQ rabbitMQ canal-server運行的模式，TCP模式就是直連客戶端，不經過中間件，kafka和mq是訊息佇列的模式
  mode: tcp 
#  flatMessage: true
  zookeeperHosts: 
  syncBatchSize: 1
  retries: 0
  timeout: 1000
  accessKey:
  secretKey:
  consumerProperties:
    # canal tcp consumer 指定canal-server的地址和埠
    canal.tcp.server.host: 127.0.0.1:11111
    canal.tcp.zookeeper.hosts: 127.0.0.1:2181
    canal.tcp.batch.size: 1
    canal.tcp.username:
    canal.tcp.password:
    

  srcDataSources: # 資料源配置，從哪里獲取資料
    defaultDS: # 指定一個名字，在ES的配置中會用到，唯一
      url: jdbc:mysql://127.0.0.1:3306/test2?useUnicode=true
      username: root
      password: *****
  canalAdapters:
  - instance: example # canal instance Name or mq topic name 指定在canal配置的實體名稱
    groups:
    - groupId: g1 
      outerAdapters:
      - name: logger
#      - name: rdb
#        key: mysql1
#        properties:
#          jdbc.driverClassName: com.mysql.jdbc.Driver
#          jdbc.url: jdbc:mysql://127.0.0.1:3306/mytest2?useUnicode=true
#          jdbc.username: root
#          jdbc.password: 121212
#      - name: rdb
#        key: oracle1
#        properties:
#          jdbc.driverClassName: oracle.jdbc.OracleDriver
#          jdbc.url: jdbc:oracle:thin:@localhost:49161:XE
#          jdbc.username: mytest
#          jdbc.password: m121212
#      - name: rdb
#        key: postgres1
#        properties:
#          jdbc.driverClassName: org.postgresql.Driver
#          jdbc.url: jdbc:postgresql://localhost:5432/postgres
#          jdbc.username: postgres
#          jdbc.password: 121212
#          threads: 1
#          commitSize: 3000
      - name: hbase
        properties:
          hbase.zookeeper.quorum: sangfor.abdi.node3,sangfor.abdi.node2,sangfor.abdi.node1
          hbase.zookeeper.property.clientPort: 2181
          zookeeper.znode.parent: /hbase-unsecure
      - name: es7 # config目錄下的子目錄名稱
        hosts: 192.168.168.2:9300 # 127.0.0.1:9200 for rest mode
        properties:
          mode: transport # or rest
#          # security.auth: test:123456 #  only used for rest mode
          cluster.name: my_application
#        - name: kudu
#          key: kudu
#          properties:
#            kudu.master.address: 127.0.0.1 # ',' split multi address

2.ElasticSearch 表映射檔案

# 指定資料源，這個值和adapter的application.yml檔案中配置的srcDataSources值對應，
dataSourceKey: defaultDS
# 指定canal-server中配置的某個實體的名字，不同實體對應不同業務
destination: example
# 組ID ，tcp方式這里填寫空，不要填寫值，不然可能會接收不到資料
groupId: 
# ES的mapping（映射）
esMapping:
  # ES索引名稱
  _index: testsync2
  # ES標示檔案的唯一標示，通常對應資料表中的主鍵ID欄位
  _id: _id
#  upsert: true
#  pk: id
# 資料表每個欄位映射到表中的具體名稱，不能重復
  sql: "select a.id as _id, a.name,a.age,a.age_2,a.message,a.insert_time from testsync as a"
#  objFields:
#    _labels: array:;
#  etlCondition: "where a.c_time>={}"
  commitBatch: 10

3 重啟服務

bin/restart.sh

寫入資料

INSERT INTO testsync(id,name,age,insert_time) values(UUID(),UUID(),2,now());
INSERT INTO testsync(id,name,age,insert_time) values(UUID(),UUID(),2,now());
INSERT INTO testsync(id,name,age,insert_time) values(UUID(),UUID(),2,now());
INSERT INTO testsync(id,name,age,insert_time) values(UUID(),UUID(),2,now());
INSERT INTO testsync(id,name,age,insert_time) values(UUID(),UUID(),2,now());
INSERT INTO testsync(id,name,age,insert_time) values(UUID(),UUID(),2,now());
INSERT INTO testsync(id,name,age,insert_time) values(UUID(),UUID(),2,now());
INSERT INTO testsync(id,name,age,insert_time) values(UUID(),UUID(),2,now());

查看adapter日志

2021-09-20 13:53:07.279 [pool-1-thread-1] INFO  c.a.o.canal.client.adapter.logger.LoggerAdapterExample - DML: {"data":[{"id":"05fabf89-19d7-11ec-bbe0-708cb6f5eaa6","name":"05fabfb4-19d7-11ec-bbe0-708cb6f5eaa6","age":2,"age_2":null,"message":null,"insert_time":1632117185000}],"database":"test2","destination":"example","es":1632117185000,"groupId":"g1","isDdl":false,"old":null,"pkNames":["id"],"sql":"","table":"testsync","ts":1632117187278,"type":"INSERT"}
2021-09-20 13:53:07.286 [pool-1-thread-1] DEBUG c.a.o.c.client.adapter.hbase.service.HbaseSyncService - DML: {"data":[{"id":"05fabf89-19d7-11ec-bbe0-708cb6f5eaa6","name":"05fabfb4-19d7-11ec-bbe0-708cb6f5eaa6","age":2,"age_2":null,"message":null,"insert_time":1632117185000}],"database":"test2","destination":"example","es":1632117185000,"groupId":"g1","isDdl":false,"old":null,"pkNames":["id"],"sql":"","table":"testsync","ts":1632117187278,"type":"INSERT"}
2021-09-20 13:53:07.287 [pool-1-thread-1] DEBUG c.a.o.canal.client.adapter.es.core.service.ESSyncService - DML: {"data":[{"id":"05fabf89-19d7-11ec-bbe0-708cb6f5eaa6","name":"05fabfb4-19d7-11ec-bbe0-708cb6f5eaa6","age":2,"age_2":null,"message":null,"insert_time":1632117185000}],"database":"test2","destination":"example","es":1632117185000,"groupId":"g1","isDdl":false,"old":null,"pkNames":["id"],"sql":"","table":"testsync","ts":1632117187278,"type":"INSERT"} 
Affected indexes: testsync2

查看ElasticSearch資料
在這里插入圖片描述
至此寫入ElasticSearch、Hbase成功

配接器映射檔案詳細介紹（單表、多表映射介紹）

${adapter}/conf/es7/xxx.yml

dataSourceKey: defaultDS        # 源資料源的key, 對應上面配置的srcDataSources中的值
outerAdapterKey: exampleKey     # 對應application.yml中es配置的key 
destination: example            # cannal的instance或者MQ的topic
groupId:                        # 對應MQ模式下的groupId, 只會同步對應groupId的資料
esMapping:
  _index: mytest_user           # es 的索引名稱
  _type: _doc                   # es 的type名稱, es7下無需配置此項
  _id: _id                      # es 的_id, 如果不配置該項必須配置下面的pk項_id則會由es自動分配
#  pk: id                       # 如果不需要_id, 則需要指定一個屬性為主鍵屬性
  # sql映射
  sql: "select a.id as _id, a.name as _name, a.role_id as _role_id, b.role_name as _role_name,
        a.c_time as _c_time, c.labels as _labels from user a
        left join role b on b.id=a.role_id
        left join (select user_id, group_concat(label order by id desc separator ';') as labels from label
        group by user_id) c on c.user_id=a.id"
#  objFields:
#    _labels: array:;           # 陣列或者物件屬性, array:; 代表以;欄位里面是以;分隔的
#    _obj: object               # json物件
  etlCondition: "where a.c_time>='{0}'"     # etl 的條件引數
  commitBatch: 3000                         # 提交批大小

sql映射說明:

sql支持多表關聯自由組合, 但是有一定的限制:

主表不能為子查詢陳述句
只能使用left outer join即最左表一定要是主表
關聯從表如果是子查詢不能有多張表
主sql中不能有where查詢條件(從表子查詢中可以有where條件但是不推薦, 可能會造成資料同步的不一致, 比如修改了where條件中的欄位內容)
關聯條件只允許主外鍵的’='操作不能出現其他常量判斷比如: on a.role_id=b.id and b.statues=1
關聯條件必須要有一個欄位出現在主查詢陳述句中比如: on a.role_id=b.id 其中的 a.role_id 或者 b.id 必須出現在主select陳述句中

Elastic Search的mapping 屬性與sql的查詢值將一一對應(不支持 select *), 比如: select a.id as _id, a.name, a.email as _email from user, 其中name將映射到es mapping的name field, _email將映射到mapping的_email field, 這里以別名(如果有別名)作為最終的映射欄位. 這里的_id可以填寫到組態檔的 _id: _id映射.

單表映射索引示例sql

select a.id as _id, a.name, a.role_id, a.c_time from user a

該sql對應的es mapping示例:

{
    "mytest_user": {
        "mappings": {
            "_doc": {
                "properties": {
                    "name": {
                        "type": "text"
                    },
                    "role_id": {
                        "type": "long"
                    },
                    "c_time": {
                        "type": "date"
                    }
                }
            }
        }
    }
}

單表映射索引示例sql帶函式或運算操作

select a.id as _id, concat(a.name,'_test') as name, a.role_id+10000 as role_id, a.c_time from user a

函式欄位后必須跟上別名, 該sql對應的es mapping示例:

{
    "mytest_user": {
        "mappings": {
            "_doc": {
                "properties": {
                    "name": {
                        "type": "text"
                    },
                    "role_id": {
                        "type": "long"
                    },
                    "c_time": {
                        "type": "date"
                    }
                }
            }
        }
    }
}

多表映射(一對一, 多對一)索引示例sql

select a.id as _id, a.name, a.role_id, b.role_name, a.c_time from user a 
left join role b on b.id = a.role_id

注:這里join操作只能是left outer join, 第一張表必須為主表!!
該sql對應的es mapping示例:

{
    "mytest_user": {
        "mappings": {
            "_doc": {
                "properties": {
                    "name": {
                        "type": "text"
                    },
                    "role_id": {
                        "type": "long"
                    },
                    "role_name": {
                        "type": "text"
                    },
                    "c_time": {
                        "type": "date"
                    }
                }
            }
        }
    }
}

多表映射(一對多)索引示例sql

select a.id as _id, a.name, a.role_id, c.labels, a.c_time from user a 
left join (select user_id, group_concat(label order by id desc separator ';') as labels from label
        group by user_id) c on c.user_id=a.id

注:left join 后的子查詢只允許一張表, 即子查詢中不能再包含子查詢或者關聯!!

該sql對應的es mapping示例:

{
    "mytest_user": {
        "mappings": {
            "_doc": {
                "properties": {
                    "name": {
                        "type": "text"
                    },
                    "role_id": {
                        "type": "long"
                    },
                    "c_time": {
                        "type": "date"
                    },
                    "labels": {
                        "type": "text"
                    }
                }
            }
        }
    }
}

其它型別的sql示例

geo type

select ... concat(IFNULL(a.latitude, 0), ',', IFNULL(a.longitude, 0)) AS location, ...

復合主鍵

select concat(a.id,'_',b.type) as _id, ... from user a left join role b on b.id=a.role_id

陣列欄位

select a.id as _id, a.name, a.role_id, c.labels, a.c_time from user a 
left join (select user_id, group_concat(label order by id desc separator ';') as labels from label
        group by user_id) c on c.user_id=a.id

配置中使用:

objFields:
  labels: array:;

物件欄位

select a.id as _id, a.name, a.role_id, c.labels, a.c_time, a.description from user a

配置中使用:

objFields:
  description: object

其中a.description欄位內容為json字串

父子檔案索引
es/customer.yml

......
esMapping:
  _index: customer
  _type: _doc
  _id: id
  relations:
    customer_order:
      name: customer
  sql: "select t.id, t.name, t.email from customer t"

es/order.yml

esMapping:
  _index: customer
  _type: _doc
  _id: _id
  relations:
    customer_order:
      name: order
      parent: customer_id
  sql: "select concat('oid_', t.id) as _id,
        t.customer_id,
        t.id as order_id,
        t.serial_code as order_serial,
        t.c_time as order_time
        from biz_order t"
  skips:
    - customer_id

mapping示例:

{
  "mappings":{
    "_doc":{
      "properties":{
        "id": {
          "type": "long"
        },
        "name": {
          "type": "text"
        },
        "email": {
          "type": "text"
        },
        "order_id": {
          "type": "long"
        },
        "order_serial": {
          "type": "text"
        },
        "order_time": {
          "type": "date"
        },
        "customer_order":{
          "type":"join",
          "relations":{
            "customer":"order"
          }
        }
      }
    }
  }
}

注意事項

多表映射時，主表資料必須插入，如果只插入子表不插入主表，資料無法同步到ElasticSearch；相反只插入主表，子表不進行插入，資料是可以同步到ElasticSearch的
多表映射時，如果主表關聯id寫入后，子表再進行修改之前的關聯的id為我們主表寫入的id，資料是無法同步到ElasticSearch中的，

轉載請註明出處，本文鏈接：https://www.uj5u.com/qita/302739.html

標籤：其他

上一篇：超詳細梳理HBase核心知識點（下）【建議收藏】

下一篇：Hadoop2.x學習筆記-1（Hadoop架構+NTP集群時間同步配置）

Canal Mysql binlog 同步至 ElasticSearch 詳細介紹