{"id":1682,"date":"2023-03-25T22:35:45","date_gmt":"2023-03-25T14:35:45","guid":{"rendered":"https:\/\/www.appblog.cn\/?p=1682"},"modified":"2023-04-23T21:48:10","modified_gmt":"2023-04-23T13:48:10","slug":"alibaba-canal-client-synchronize-elasticsearch","status":"publish","type":"post","link":"https:\/\/www.appblog.cn\/index.php\/2023\/03\/25\/alibaba-canal-client-synchronize-elasticsearch\/","title":{"rendered":"\u963f\u91cc\u5df4\u5df4Canal Client\u540c\u6b65ElasticSearch"},"content":{"rendered":"

\u53c2\u8003\uff1ahttps:\/\/github.com\/alibaba\/canal\/wiki\/Sync-ES<\/a><\/p>\n

\u80cc\u666f<\/h2>\n

canal 1.1.1\u7248\u672c\u4e4b\u540e\uff0c\u5185\u7f6e\u589e\u52a0\u5ba2\u6237\u7aef\u6570\u636e\u540c\u6b65\u529f\u80fd, Client\u9002\u914d\u5668\u6574\u4f53\u4ecb\u7ecd\uff1aClientAdapter<\/a><\/p>\n

canal adapter<\/code>\u7684 Elastic Search \u7248\u672c\u652f\u63016.x.x<\/code>\u4ee5\u4e0a, \u5982\u9700\u5176\u5b83\u7248\u672c\u7684es\u53ef\u66ff\u6362\u4f9d\u8d56\u91cd\u65b0\u7f16\u8bd1client-adapter.elasticsearch<\/code>\u6a21\u5757<\/p>\n

<\/p>\n

ElasticSearch\u9002\u914d\u5668<\/h2>\n

\u4fee\u6539\u542f\u52a8\u5668\u914d\u7f6e: application.yml<\/h3>\n
canal.conf:\n  canalServerHost: 127.0.0.1:11111\n  batchSize: 500\n  syncBatchSize: 1000\n  retries: 0\n  timeout:\n  mode: tcp \n  srcDataSources:\n    defaultDS:\n      url: jdbc:mysql:\/\/127.0.0.1:3306\/mytest?useUnicode=true\n      username: root\n      password: 121212\n  canalAdapters:\n  - instance: example \n    groups:\n    - groupId: g1\n      outerAdapters:\n      - \n        key: exampleKey\n        name: es6                           # or es7\n        hosts: 127.0.0.1:9300               # es \u96c6\u7fa4\u5730\u5740, \u9017\u53f7\u5206\u9694\n        properties:\n          mode: transport # or rest         # \u53ef\u6307\u5b9atransport\u6a21\u5f0f\u6216\u8005rest\u6a21\u5f0f\n          # security.auth: test:123456      # only used for rest mode\n          cluster.name: elasticsearch       # es cluster name<\/code><\/pre>\n
adapter\u5c06\u4f1a\u81ea\u52a8\u52a0\u8f7dconf\/es<\/code>\u4e0b\u7684\u6240\u6709.yml<\/code>\u7ed3\u5c3e\u7684\u914d\u7f6e\u6587\u4ef6<\/p>\n
\u9002\u914d\u5668\u8868\u6620\u5c04\u6587\u4ef6<\/h3>\n
\u4fee\u6539conf\/es\/mytest_user.yml<\/code>\u6587\u4ef6:<\/p>\n
dataSourceKey: defaultDS        # \u6e90\u6570\u636e\u6e90\u7684key, \u5bf9\u5e94\u4e0a\u9762\u914d\u7f6e\u7684srcDataSources\u4e2d\u7684\u503c\nouterAdapterKey: exampleKey     # \u5bf9\u5e94application.yml\u4e2des\u914d\u7f6e\u7684key \ndestination: example            # cannal\u7684instance\u6216\u8005MQ\u7684topic\ngroupId:                        # \u5bf9\u5e94MQ\u6a21\u5f0f\u4e0b\u7684groupId, \u53ea\u4f1a\u540c\u6b65\u5bf9\u5e94groupId\u7684\u6570\u636e\nesMapping:\n  _index: mytest_user           # es \u7684\u7d22\u5f15\u540d\u79f0\n  _type: _doc                   # es \u7684type\u540d\u79f0, es7\u4e0b\u65e0\u9700\u914d\u7f6e\u6b64\u9879\n  _id: _id                      # es \u7684_id, \u5982\u679c\u4e0d\u914d\u7f6e\u8be5\u9879\u5fc5\u987b\u914d\u7f6e\u4e0b\u9762\u7684pk\u9879_id\u5219\u4f1a\u7531es\u81ea\u52a8\u5206\u914d\n#  pk: id                       # \u5982\u679c\u4e0d\u9700\u8981_id, \u5219\u9700\u8981\u6307\u5b9a\u4e00\u4e2a\u5c5e\u6027\u4e3a\u4e3b\u952e\u5c5e\u6027\n  # sql\u6620\u5c04\n  sql: "select a.id as _id, a.name as _name, a.role_id as _role_id, b.role_name as _role_name,\n        a.c_time as _c_time, c.labels as _labels from user a\n        left join role b on b.id=a.role_id\n        left join (select user_id, group_concat(label order by id desc separator ';') as labels from label\n        group by user_id) c on c.user_id=a.id"\n#  objFields:\n#    _labels: array:;           # \u6570\u7ec4\u6216\u8005\u5bf9\u8c61\u5c5e\u6027, array:; \u4ee3\u8868\u4ee5;\u5b57\u6bb5\u91cc\u9762\u662f\u4ee5;\u5206\u9694\u7684\n#    _obj: object               # json\u5bf9\u8c61\n  etlCondition: "where a.c_time>='{0}'"     # etl \u7684\u6761\u4ef6\u53c2\u6570\n  commitBatch: 3000                         # \u63d0\u4ea4\u6279\u5927\u5c0f<\/code><\/pre>\n
sql\u6620\u5c04\u8bf4\u660e<\/h4>\n
sql\u652f\u6301\u591a\u8868\u5173\u8054\u81ea\u7531\u7ec4\u5408\uff0c\u4f46\u662f\u6709\u4e00\u5b9a\u7684\u9650\u5236:<\/p>\n
    \n
  • \u4e3b\u8868\u4e0d\u80fd\u4e3a\u5b50\u67e5\u8be2\u8bed\u53e5<\/li>\n
  • \u53ea\u80fd\u4f7f\u7528left outer join<\/code>\uff0c\u5373\u6700\u5de6\u8868\u4e00\u5b9a\u8981\u662f\u4e3b\u8868<\/li>\n
  • \u5173\u8054\u4ece\u8868\u5982\u679c\u662f\u5b50\u67e5\u8be2\u4e0d\u80fd\u6709\u591a\u5f20\u8868<\/li>\n
  • \u4e3bsql\u4e2d\u4e0d\u80fd\u6709where\u67e5\u8be2\u6761\u4ef6(\u4ece\u8868\u5b50\u67e5\u8be2\u4e2d\u53ef\u4ee5\u6709where\u6761\u4ef6\u4f46\u662f\u4e0d\u63a8\u8350\uff0c\u53ef\u80fd\u4f1a\u9020\u6210\u6570\u636e\u540c\u6b65\u7684\u4e0d\u4e00\u81f4\uff0c\u6bd4\u5982\u4fee\u6539\u4e86where\u6761\u4ef6\u4e2d\u7684\u5b57\u6bb5\u5185\u5bb9)<\/li>\n
  • \u5173\u8054\u6761\u4ef6\u53ea\u5141\u8bb8\u4e3b\u5916\u952e\u7684=<\/code>\u64cd\u4f5c\u4e0d\u80fd\u51fa\u73b0\u5176\u4ed6\u5e38\u91cf\u5224\u65ad\u6bd4\u5982: on a.role_id=b.id and b.statues=1<\/code><\/li>\n
  • \u5173\u8054\u6761\u4ef6\u5fc5\u987b\u8981\u6709\u4e00\u4e2a\u5b57\u6bb5\u51fa\u73b0\u5728\u4e3b\u67e5\u8be2\u8bed\u53e5\u4e2d\u6bd4\u5982: on a.role_id=b.id<\/code>\uff0c\u5176\u4e2d\u7684a.role_id<\/code>\u6216\u8005b.id<\/code>\u5fc5\u987b\u51fa\u73b0\u5728\u4e3bselect\u8bed\u53e5\u4e2d<\/li>\n<\/ul>\n
    ElasticSearch\u7684mapping \u5c5e\u6027\u4e0esql\u7684\u67e5\u8be2\u503c\u5c06\u4e00\u4e00\u5bf9\u5e94(\u4e0d\u652f\u6301select *<\/code>), \u6bd4\u5982\uff1aselect a.id as _id, a.name, a.email as _email from user<\/code>\uff0c\u5176\u4e2dname<\/code>\u5c06\u6620\u5c04\u5230es mapping\u7684name field<\/code>\uff0c_email<\/code>\u5c06 \u6620\u5c04\u5230mapping\u7684_email field<\/code>\uff0c\u8fd9\u91cc\u4ee5\u522b\u540d(\u5982\u679c\u6709\u522b\u540d)\u4f5c\u4e3a\u6700\u7ec8\u7684\u6620\u5c04\u5b57\u6bb5\u3002\u8fd9\u91cc\u7684_id<\/code>\u53ef\u4ee5\u586b\u5199\u5230\u914d\u7f6e\u6587\u4ef6\u7684_id: _id<\/code>\u6620\u5c04\u3002<\/p>\n
    \u5355\u8868\u6620\u5c04\u7d22\u5f15\u793a\u4f8bsql<\/h4>\n
    select a.id as _id, a.name, a.role_id, a.c_time from user a<\/code><\/pre>\n
    \u8be5sql\u5bf9\u5e94\u7684es mapping\u793a\u4f8b\uff1a<\/p>\n
    {\n    "mytest_user": {\n        "mappings": {\n            "_doc": {\n                "properties": {\n                    "name": {\n                        "type": "text"\n                    },\n                    "role_id": {\n                        "type": "long"\n                    },\n                    "c_time": {\n                        "type": "date"\n                    }\n                }\n            }\n        }\n    }\n}<\/code><\/pre>\n
    \u5355\u8868\u6620\u5c04\u7d22\u5f15\u793a\u4f8bsql\u5e26\u51fd\u6570\u6216\u8fd0\u7b97\u64cd\u4f5c<\/h4>\n
    select a.id as _id, concat(a.name,'_test') as name, a.role_id+10000 as role_id, a.c_time from user a<\/code><\/pre>\n
    \u51fd\u6570\u5b57\u6bb5\u540e\u5fc5\u987b\u8ddf\u4e0a\u522b\u540d\uff0c\u8be5sql\u5bf9\u5e94\u7684es mapping\u793a\u4f8b\uff1a<\/p>\n
    {\n    "mytest_user": {\n        "mappings": {\n            "_doc": {\n                "properties": {\n                    "name": {\n                        "type": "text"\n                    },\n                    "role_id": {\n                        "type": "long"\n                    },\n                    "c_time": {\n                        "type": "date"\n                    }\n                }\n            }\n        }\n    }\n}<\/code><\/pre>\n
    \u591a\u8868\u6620\u5c04(\u4e00\u5bf9\u4e00\uff0c\u591a\u5bf9\u4e00)\u7d22\u5f15\u793a\u4f8bsql<\/h4>\n
    select a.id as _id, a.name, a.role_id, b.role_name, a.c_time from user a \nleft join role b on b.id = a.role_id<\/code><\/pre>\n
    \n
    \u6ce8\uff1a\u8fd9\u91ccjoin\u64cd\u4f5c\u53ea\u80fd\u662fleft outer join<\/code>\uff0c\u7b2c\u4e00\u5f20\u8868\u5fc5\u987b\u4e3a\u4e3b\u8868!!<\/p>\n<\/blockquote>\n
    \u8be5sql\u5bf9\u5e94\u7684es mapping\u793a\u4f8b\uff1a<\/p>\n
    {\n    "mytest_user": {\n        "mappings": {\n            "_doc": {\n                "properties": {\n                    "name": {\n                        "type": "text"\n                    },\n                    "role_id": {\n                        "type": "long"\n                    },\n                    "role_name": {\n                        "type": "text"\n                    },\n                    "c_time": {\n                        "type": "date"\n                    }\n                }\n            }\n        }\n    }\n}<\/code><\/pre>\n
    \u591a\u8868\u6620\u5c04(\u4e00\u5bf9\u591a)\u7d22\u5f15\u793a\u4f8bsql<\/h4>\n
    select a.id as _id, a.name, a.role_id, c.labels, a.c_time from user a \nleft join (select user_id, group_concat(label order by id desc separator ';') as labels from label\n        group by user_id) c on c.user_id=a.id<\/code><\/pre>\n
    \n
    \u6ce8\uff1aleft join<\/code>\u540e\u7684\u5b50\u67e5\u8be2\u53ea\u5141\u8bb8\u4e00\u5f20\u8868\uff0c\u5373\u5b50\u67e5\u8be2\u4e2d\u4e0d\u80fd\u518d\u5305\u542b\u5b50\u67e5\u8be2\u6216\u8005\u5173\u8054!!<\/p>\n<\/blockquote>\n
    \u8be5sql\u5bf9\u5e94\u7684es mapping\u793a\u4f8b\uff1a<\/p>\n
    {\n    "mytest_user": {\n        "mappings": {\n            "_doc": {\n                "properties": {\n                    "name": {\n                        "type": "text"\n                    },\n                    "role_id": {\n                        "type": "long"\n                    },\n                    "c_time": {\n                        "type": "date"\n                    },\n                    "labels": {\n                        "type": "text"\n                    }\n                }\n            }\n        }\n    }\n}<\/code><\/pre>\n
    \u5176\u5b83\u7c7b\u578b\u7684sql\u793a\u4f8b<\/h4>\n
    \uff081\uff09geo type<\/p>\n
    select ... concat(IFNULL(a.latitude, 0), ',', IFNULL(a.longitude, 0)) AS location, ...<\/code><\/pre>\n
    \uff082\uff09\u590d\u5408\u4e3b\u952e<\/p>\n
    select concat(a.id,'_',b.type) as _id, ... from user a left join role b on b.id=a.role_id<\/code><\/pre>\n
    \uff083\uff09\u6570\u7ec4\u5b57\u6bb5<\/p>\n
    select a.id as _id, a.name, a.role_id, c.labels, a.c_time from user a \nleft join (select user_id, group_concat(label order by id desc separator ';') as labels from label\n        group by user_id) c on c.user_id=a.id<\/code><\/pre>\n
    \u914d\u7f6e\u4e2d\u4f7f\u7528\uff1a<\/p>\n
    objFields:\n  labels: array:;<\/code><\/pre>\n
    \uff084\uff09\u5bf9\u8c61\u5b57\u6bb5<\/p>\n
    select a.id as _id, a.name, a.role_id, c.labels, a.c_time, a.description from user a<\/code><\/pre>\n
    \u914d\u7f6e\u4e2d\u4f7f\u7528\uff1a<\/p>\n
    objFields:\n  description: object<\/code><\/pre>\n
    \u5176\u4e2da.description<\/code>\u5b57\u6bb5\u5185\u5bb9\u4e3ajson\u5b57\u7b26\u4e32<\/p>\n
    \uff085\uff09\u7236\u5b50\u6587\u6863\u7d22\u5f15<\/p>\n
    es\/customer.yml<\/p>\n
    esMapping:\n  _index: customer\n  _type: _doc\n  _id: id\n  relations:\n    customer_order:\n      name: customer\n  sql: "select t.id, t.name, t.email from customer t"<\/code><\/pre>\n
    es\/order.yml<\/p>\n
    esMapping:\n  _index: customer\n  _type: _doc\n  _id: _id\n  relations:\n    customer_order:\n      name: order\n      parent: customer_id\n  sql: "select concat('oid_', t.id) as _id,\n        t.customer_id,\n        t.id as order_id,\n        t.serial_code as order_serial,\n        t.c_time as order_time\n        from biz_order t"\n  skips:\n    - customer_id<\/code><\/pre>\n
    mapping\u793a\u4f8b\uff1a<\/p>\n
    {\n  "mappings":{\n    "_doc":{\n      "properties":{\n        "id": {\n          "type": "long"\n        },\n        "name": {\n          "type": "text"\n        },\n        "email": {\n          "type": "text"\n        },\n        "order_id": {\n          "type": "long"\n        },\n        "order_serial": {\n          "type": "text"\n        },\n        "order_time": {\n          "type": "date"\n        },\n        "customer_order":{\n          "type":"join",\n          "relations":{\n            "customer":"order"\n          }\n        }\n      }\n    }\n  }\n}<\/code><\/pre>\n
    \u542f\u52a8ES\u6570\u636e\u540c\u6b65<\/h3>\n
    \u542f\u52a8canal-adapter\u542f\u52a8\u5668<\/strong><\/p>\n
    bin\/startup.sh<\/code><\/pre>\n
    \u9a8c\u8bc1<\/strong><\/p>\n
      \n
    • \u65b0\u589emysql mytest.user<\/code>\u8868\u7684\u6570\u636e\uff0c\u5c06\u4f1a\u81ea\u52a8\u540c\u6b65\u5230es\u7684mytest_user<\/code>\u7d22\u5f15\u4e0b\u9762, \u5e76\u4f1a\u6253\u51faDML\u7684log<\/li>\n
    • \u4fee\u6539mysql mytest.role<\/code>\u8868\u7684role_name\uff0c\u5c06\u4f1a\u81ea\u52a8\u540c\u6b65es\u7684mytest_user<\/code>\u7d22\u5f15\u4e2d\u7684role_name\u6570\u636e<\/li>\n
    • \u65b0\u589e\u6216\u8005\u4fee\u6539mysql mytest.label<\/code>\u8868\u7684label\uff0c\u5c06\u4f1a\u81ea\u52a8\u540c\u6b65es\u7684mytest_user<\/code>\u7d22\u5f15\u4e2d\u7684labels\u6570\u636e<\/li>\n<\/ul>\n","protected":false},"excerpt":{"rendered":"
      \u53c2\u8003\uff1ahttps:\/\/github.com\/alibaba\/canal\/wiki\/Sync-ES \u80cc\u666f can […]<\/p>\n","protected":false},"author":1,"featured_media":0,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[417],"tags":[],"class_list":["post-1682","post","type-post","status-publish","format-standard","hentry","category-canal"],"_links":{"self":[{"href":"https:\/\/www.appblog.cn\/index.php\/wp-json\/wp\/v2\/posts\/1682","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/www.appblog.cn\/index.php\/wp-json\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/www.appblog.cn\/index.php\/wp-json\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/www.appblog.cn\/index.php\/wp-json\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/www.appblog.cn\/index.php\/wp-json\/wp\/v2\/comments?post=1682"}],"version-history":[{"count":0,"href":"https:\/\/www.appblog.cn\/index.php\/wp-json\/wp\/v2\/posts\/1682\/revisions"}],"wp:attachment":[{"href":"https:\/\/www.appblog.cn\/index.php\/wp-json\/wp\/v2\/media?parent=1682"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/www.appblog.cn\/index.php\/wp-json\/wp\/v2\/categories?post=1682"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/www.appblog.cn\/index.php\/wp-json\/wp\/v2\/tags?post=1682"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}