{"id":217480,"date":"2020-12-04T21:31:25","date_gmt":"2020-12-04T13:31:25","guid":{"rendered":"http:\/\/4563.org\/?p=217480"},"modified":"2020-12-04T21:31:25","modified_gmt":"2020-12-04T13:31:25","slug":"nebula-flink-connector-%e7%9a%84%e5%8e%9f%e7%90%86%e5%92%8c%e5%ae%9e%e8%b7%b5","status":"publish","type":"post","link":"http:\/\/4563.org\/?p=217480","title":{"rendered":"Nebula Flink Connector \u7684\u539f\u7406\u548c\u5b9e\u8df5"},"content":{"rendered":"
\n
\n
\n

Nebula Flink Connector \u7684\u539f\u7406\u548c\u5b9e\u8df5 <\/h1>\n

<\/p>\n

\n
\u8cc7\u6df1\u5927\u4f6c : NebulaGraph <\/span> <\/i> 2<\/span> <\/div>\n
<\/div>\n<\/p><\/div>\n<\/p><\/div>\n<\/p><\/div>\n
<\/p>\n

\u6458\u8981\uff1a\u672c\u6587\u6240\u4ecb\u7ecd Nebula Graph \u8fde\u63a5\u5668 Nebula Flink Connector\uff0c\u91c7\u7528\u7c7b\u4f3c Flink \u63d0\u4f9b\u7684 Flink Connector \u5f62\u5f0f\uff0c\u652f\u6301 Flink \u8bfb\u5199\u5206\u5e03\u5f0f\u56fe\u6570\u636e\u5e93 Nebula Graph \u3002<\/p>\n

\n

\u6587\u7ae0\u9996\u53d1 Nebula Graph \u5b98\u7f51\u535a\u5ba2\uff1ahttps:\/\/nebula-graph.com.cn\/posts\/nebula-flink-connector\/<\/p>\n<\/blockquote>\n

\"Nebula<\/p>\n

\u5728\u5173\u7cfb\u7f51\u7edc\u5206\u6790\u3001\u5173\u7cfb\u5efa\u6a21\u3001\u5b9e\u65f6\u63a8\u8350\u7b49\u573a\u666f\u4e2d\u5e94\u7528\u56fe\u6570\u636e\u5e93\u4f5c\u4e3a\u540e\u53f0\u6570\u636e\u652f\u6491\u5df2\u76f8\u5bf9\u666e\u53ca\uff0c\u4e14\u90e8\u5206\u5e94\u7528\u573a\u666f\u5bf9\u56fe\u6570\u636e\u7684\u5b9e\u65f6\u6027\u8981\u6c42\u8f83\u9ad8\uff0c\u5982\u63a8\u8350\u7cfb\u7edf\u3001\u641c\u7d22\u5f15\u64ce\u3002\u4e3a\u4e86\u63d0\u5347\u6570\u636e\u7684\u5b9e\u65f6\u6027\uff0c\u4e1a\u754c\u5e7f\u6cdb\u5e94\u7528\u6d41\u5f0f\u8ba1\u7b97\u5bf9\u66f4\u65b0\u7684\u6570\u636e\u8fdb\u884c\u589e\u91cf\u5b9e\u65f6\u5904\u7406\u3002\u4e3a\u4e86\u652f\u6301\u5bf9\u56fe\u6570\u636e\u7684\u6d41\u5f0f\u8ba1\u7b97\uff0cNebula Graph \u56e2\u961f\u5f00\u53d1\u4e86 Nebula Flink Connector\uff0c\u652f\u6301\u5229\u7528 Flink \u8fdb\u884c Nebula Graph \u56fe\u6570\u636e\u7684\u6d41\u5f0f\u5904\u7406\u548c\u8ba1\u7b97\u3002<\/p>\n

Flink \u662f\u65b0\u4e00\u4ee3\u6d41\u6279\u7edf\u4e00\u7684\u8ba1\u7b97\u5f15\u64ce\uff0c\u5b83\u4ece\u4e0d\u540c\u7684\u7b2c\u4e09\u65b9\u5b58\u50a8\u5f15\u64ce\u4e2d\u8bfb\u53d6\u6570\u636e\uff0c\u5e76\u8fdb\u884c\u5904\u7406\uff0c\u518d\u5199\u5165\u53e6\u5916\u7684\u5b58\u50a8\u5f15\u64ce\u4e2d\u3002Flink Connector<\/strong> \u7684\u4f5c\u7528\u5c31\u76f8\u5f53\u4e8e\u4e00\u4e2a\u8fde\u63a5\u5668\uff0c\u8fde\u63a5 Flink<\/strong> \u8ba1\u7b97\u5f15\u64ce\u8ddf\u5916\u754c\u5b58\u50a8\u7cfb\u7edf\u3002<\/p>\n

\u4e0e\u5916\u754c\u8fdb\u884c\u6570\u636e\u4ea4\u6362\u65f6\uff0cFlink \u652f\u6301\u4ee5\u4e0b 4 \u79cd\u65b9\u5f0f\uff1a<\/p>\n

    \n
  • Flink \u6e90\u7801\u5185\u90e8\u9884\u5b9a\u4e49 Source \u548c Sink \u7684 API \uff1b<\/li>\n
  • Flink \u5185\u90e8\u63d0\u4f9b\u4e86 Bundled Connectors\uff0c\u5982 JDBC Connector \u3002<\/li>\n
  • Apache Bahir \u9879\u76ee\u4e2d\u63d0\u4f9b\u8fde\u63a5\u5668 Apache Bahir \u6700\u521d\u662f\u4ece Apache Spark \u4e2d\u72ec\u7acb\u51fa\u6765\u7684\u9879\u76ee\uff0c\u4ee5\u63d0\u4f9b\u4e0d\u9650\u4e8e Spark \u76f8\u5173\u7684\u6269\u5c55 \/\u63d2\u4ef6\u3001\u8fde\u63a5\u5668\u548c\u5176\u4ed6\u53ef\u63d2\u5165\u7ec4\u4ef6\u7684\u5b9e\u73b0\u3002<\/li>\n
  • \u901a\u8fc7\u5f02\u6b65 I\/O \u65b9\u5f0f\u3002<\/li>\n<\/ul>\n

    \u6d41\u8ba1\u7b97\u4e2d\u7ecf\u5e38\u9700\u8981\u4e0e\u5916\u90e8\u5b58\u50a8\u7cfb\u7edf\u4ea4\u4e92\uff0c\u6bd4\u5982\u9700\u8981\u5173\u8054 MySQL \u4e2d\u7684\u67d0\u4e2a\u8868\u3002\u4e00\u822c\u6765\u8bf4\uff0c\u5982\u679c\u7528\u540c\u6b65 I\/O \u7684\u65b9\u5f0f\uff0c\u4f1a\u9020\u6210\u7cfb\u7edf\u4e2d\u51fa\u73b0\u5927\u7684\u7b49\u5f85\u65f6\u95f4\uff0c\u5f71\u54cd\u541e\u5410\u548c\u5ef6\u8fdf\u3002\u5f02\u6b65 I\/O \u5219\u53ef\u4ee5\u5e76\u53d1\u5904\u7406\u591a\u4e2a\u8bf7\u6c42\uff0c\u63d0\u9ad8\u541e\u5410\uff0c\u51cf\u5c11\u5ef6\u8fdf\u3002<\/p>\n

    \u672c\u6587\u6240\u4ecb\u7ecd Nebula Graph \u8fde\u63a5\u5668 Nebula Flink Connector\uff0c\u91c7\u7528\u7c7b\u4f3c Flink \u63d0\u4f9b\u7684 Flink Connector \u5f62\u5f0f\uff0c\u652f\u6301 Flink \u8bfb\u5199\u5206\u5e03\u5f0f\u56fe\u6570\u636e\u5e93 Nebula Graph \u3002<\/p>\n

    \u4e00\u3001Connector Source<\/h2>\n

    Flink \u4f5c\u4e3a\u4e00\u6b3e\u6d41\u5f0f\u8ba1\u7b97\u6846\u67b6\uff0c\u5b83\u53ef\u5904\u7406\u6709\u754c\u6570\u636e\uff0c\u4e5f\u53ef\u5904\u7406\u65e0\u754c\u6570\u636e\u3002\u6240\u8c13\u65e0\u754c\uff0c\u5373\u6e90\u6e90\u4e0d\u65ad\u7684\u6570\u636e\uff0c\u4e0d\u4f1a\u6709\u7ec8\u6b62\uff0c\u5b9e\u65f6\u6d41\u5904\u7406\u6240\u5904\u7406\u7684\u6570\u636e\u4fbf\u662f\u65e0\u754c\u6570\u636e\uff1b\u6279\u5904\u7406\u7684\u6570\u636e\uff0c\u5373\u6709\u754c\u6570\u636e\u3002\u800c Source \u4fbf\u662f Flink \u5904\u7406\u6570\u636e\u7684\u6570\u636e\u6765\u6e90\u3002<\/p>\n

    Nebula Flink Connector \u4e2d\u7684 Source \u4fbf\u662f\u56fe\u6570\u636e\u5e93 Nebula Graph \u3002Flink \u63d0\u4f9b\u4e86\u4e30\u5bcc\u7684 Connector \u7ec4\u4ef6\u5141\u8bb8\u7528\u6237\u81ea\u5b9a\u4e49\u6570\u636e\u6e90\u6765\u8fde\u63a5\u5916\u90e8\u6570\u636e\u5b58\u50a8\u7cfb\u7edf\u3002<\/p>\n

    1.1 Source \u7b80\u4ecb<\/h3>\n

    Flink \u7684 Source \u4e3b\u8981\u8d1f\u8d23\u5916\u90e8\u6570\u636e\u6e90\u7684\u63a5\u5165\uff0cFlink \u7684 Source \u80fd\u529b\u4e3b\u8981\u662f\u901a\u8fc7 read \u76f8\u5173\u7684 API \u548c addSource \u65b9\u6cd5\u8fd9 2 \u79cd\u65b9\u5f0f\u6765\u5b9e\u73b0\u6570\u636e\u6e90\u7684\u8bfb\u53d6\uff0c\u4f7f\u7528 addSource \u65b9\u6cd5\u5bf9\u63a5\u5916\u90e8\u6570\u636e\u6e90\u65f6\uff0c\u53ef\u4ee5\u4f7f\u7528 Flink Bundled Connector\uff0c\u4e5f\u53ef\u4ee5\u81ea\u5b9a\u4e49 Source \u3002<\/p>\n

    Flink Source \u7684\u51e0\u79cd\u4f7f\u7528\u65b9\u5f0f\u5982\u4e0b\uff1a<\/p>\n

    \"Nebula<\/p>\n

    \u672c\u7ae0\u4e3b\u8981\u4ecb\u7ecd\u5982\u4f55\u901a\u8fc7\u81ea\u5b9a\u4e49 Source \u65b9\u5f0f\u5b9e\u73b0 Nebula Graph Source \u3002<\/p>\n

    1.2 \u81ea\u5b9a\u4e49 Source<\/h3>\n

    \u5728 Flink \u4e2d\u53ef\u4ee5\u4f7f\u7528 StreamExecutionEnvironment.addSource(sourceFunction)<\/code> \u548c ExecutionEnvironment.createInput(inputFormat)<\/code> \u4e24\u79cd\u65b9\u5f0f\u6765\u4e3a\u4f60\u7684\u7a0b\u5e8f\u6dfb\u52a0\u6570\u636e\u6765\u6e90\u3002<\/p>\n

    Flink \u5df2\u7ecf\u63d0\u4f9b\u591a\u4e2a\u5185\u7f6e\u7684 source functions<\/code> \uff0c\u5f00\u53d1\u8005\u53ef\u4ee5\u901a\u8fc7\u7ee7\u627f RichSourceFunction<\/code>\u6765\u81ea\u5b9a\u4e49\u975e\u5e76\u884c\u7684 source<\/code> \uff0c\u901a\u8fc7\u7ee7\u627f RichParallelSourceFunction<\/code> \u6765\u81ea\u5b9a\u4e49\u5e76\u884c\u7684 Source<\/code> \u3002RichSourceFunction<\/code>\u00a0\u548c RichParallelSourceFunction<\/code>\u00a0\u662f SourceFunction<\/code>\u00a0\u548c RichFunction<\/code>\u00a0\u7279\u6027\u7684\u7ed3\u5408\u3002 \u5176\u4e2dSourceFunction<\/code>\u00a0\u8d1f\u8d23\u6570\u636e\u7684\u751f\u6210\uff0c RichFunction<\/code>\u00a0\u8d1f\u8d23\u8d44\u6e90\u7684\u7ba1\u7406\u3002\u5f53\u7136\uff0c\u4e5f\u53ef\u4ee5\u53ea\u5b9e\u73b0 SourceFunction<\/code>\u00a0\u63a5\u53e3\u6765\u5b9a\u4e49\u6700\u7b80\u5355\u7684\u53ea\u5177\u5907\u83b7\u53d6\u6570\u636e\u529f\u80fd\u7684 dataSource<\/code>\u00a0\u3002<\/p>\n

    \u901a\u5e38\u81ea\u5b9a\u4e49\u4e00\u4e2a\u5b8c\u5584\u7684 Source \u8282\u70b9\u662f\u901a\u8fc7\u5b9e\u73b0 RichSourceFunction<\/code> \u7c7b\u6765\u5b8c\u6210\u7684\uff0c\u8be5\u7c7b\u517c\u5177 RichFunction<\/code> \u548c SourceFunction<\/code> \u7684\u80fd\u529b\uff0c\u56e0\u6b64\u81ea\u5b9a\u4e49 Flink \u7684 Nebula Graph Source \u529f\u80fd\u6211\u4eec\u9700\u8981\u5b9e\u73b0 RichSourceFunction<\/code> \u4e2d\u63d0\u4f9b\u7684\u65b9\u6cd5\u3002<\/p>\n

    1.3 \u81ea\u5b9a\u4e49 Nebula Graph Source \u5b9e\u73b0\u539f\u7406<\/h3>\n

    Nebula Flink Connector \u4e2d\u5b9e\u73b0\u7684\u81ea\u5b9a\u4e49 Nebula Graph Source \u6570\u636e\u6e90\u63d0\u4f9b\u4e86\u4e24\u79cd\u4f7f\u7528\u65b9\u5f0f\uff0c\u5206\u522b\u662f addSource \u548c createInput \u65b9\u5f0f\u3002<\/p>\n

    Nebula Graph Source \u5b9e\u73b0\u7c7b\u56fe\u5982\u4e0b\uff1a<\/p>\n

    \"Nebula<\/p>\n

    \uff08 1 \uff09 addSource<\/h4>\n

    \u8be5\u65b9\u5f0f\u662f\u901a\u8fc7 NebulaSourceFunction \u7c7b\u5b9e\u73b0\u7684\uff0c\u8be5\u7c7b\u7ee7\u627f\u81ea RichSourceFunction \u5e76\u5b9e\u73b0\u4e86\u4ee5\u4e0b\u65b9\u6cd5\uff1a<\/p>\n

      \n
    • open \u51c6\u5907 Nebula Graph \u8fde\u63a5\u4fe1\u606f\uff0c\u5e76\u83b7\u53d6 Nebula Graph Meta \u670d\u52a1\u548c Storage \u670d\u52a1\u7684\u8fde\u63a5\u3002<\/li>\n
    • close \u6570\u636e\u8bfb\u53d6\u5b8c\u6210\uff0c\u91ca\u653e\u8d44\u6e90\u3002\u5173\u95ed Nebula Graph \u670d\u52a1\u7684\u8fde\u63a5\u3002<\/li>\n
    • run \u5f00\u59cb\u8bfb\u53d6\u6570\u636e\uff0c\u5e76\u5c06\u6570\u636e\u586b\u5145\u5230 sourceContext \u3002<\/li>\n
    • cancel \u53d6\u6d88 Flink \u4f5c\u4e1a\u65f6\u8c03\u7528\uff0c\u5173\u95ed\u8d44\u6e90\u3002<\/li>\n<\/ul>\n

      \uff08 2 \uff09 createInput<\/h4>\n

      \u8be5\u65b9\u5f0f\u662f\u901a\u8fc7 NebulaInputFormat \u7c7b\u5b9e\u73b0\u7684\uff0c\u8be5\u7c7b\u7ee7\u627f\u81ea RichInputFormat \u5e76\u5b9e\u73b0\u4e86\u4ee5\u4e0b\u65b9\u6cd5\uff1a<\/p>\n

        \n
      • openInputFormat \u51c6\u5907 inputFormat\uff0c\u83b7\u53d6\u8fde\u63a5\u3002<\/li>\n
      • closeInputFormat \u6570\u636e\u8bfb\u53d6\u5b8c\u6210\uff0c\u91ca\u653e\u8d44\u6e90\uff0c\u5173\u95ed Nebula Graph \u670d\u52a1\u7684\u8fde\u63a5\u3002<\/li>\n
      • getStatistics \u83b7\u53d6\u6570\u636e\u6e90\u7684\u57fa\u672c\u7edf\u8ba1\u4fe1\u606f\u3002<\/li>\n
      • createInputSplits \u57fa\u4e8e\u914d\u7f6e\u7684 partition \u53c2\u6570\u521b\u5efa GenericInputSplit \u3002<\/li>\n
      • getInputSplitAssigner \u8fd4\u56de\u8f93\u5165\u7684 split \u5206\u914d\u5668\uff0c\u6309\u539f\u59cb\u8ba1\u7b97\u7684\u987a\u5e8f\u8fd4\u56de Source \u7684\u6240\u6709 split \u3002<\/li>\n
      • open \u5f00\u59cb inputFormat \u7684\u6570\u636e\u8bfb\u53d6\uff0c\u5c06\u8bfb\u53d6\u7684\u6570\u636e\u8f6c\u6362 Flink \u7684\u6570\u636e\u683c\u5f0f\uff0c\u6784\u9020\u8fed\u4ee3\u5668\u3002<\/li>\n
      • close \u6570\u636e\u8bfb\u53d6\u5b8c\u6210\uff0c\u6253\u5370\u8bfb\u53d6\u65e5\u5fd7\u3002<\/li>\n
      • reachedEnd \u662f\u5426\u8bfb\u53d6\u5b8c\u6210<\/li>\n
      • nextRecord \u901a\u8fc7\u8fed\u4ee3\u5668\u83b7\u53d6\u4e0b\u4e00\u6761\u6570\u636e<\/li>\n<\/ul>\n

        \u901a\u8fc7 addSource \u8bfb\u53d6 Source \u6570\u636e\u5f97\u5230\u7684\u662f Flink \u7684 DataStreamSource\uff0c\u8868\u793a DataStream \u7684\u8d77\u70b9\u3002<\/p>\n

        \u901a\u8fc7 createInput \u8bfb\u53d6\u6570\u636e\u5f97\u5230\u7684\u662f Flink \u7684 DataSource\uff0cDataSource \u662f\u4e00\u4e2a\u521b\u5efa\u65b0\u6570\u636e\u96c6\u7684 Operator\uff0c\u8fd9\u4e2a Operator \u53ef\u4f5c\u4e3a\u8fdb\u4e00\u6b65\u8f6c\u6362\u7684\u6570\u636e\u96c6\u3002DataSource \u53ef\u4ee5\u901a\u8fc7 withParameters \u5c01\u88c5\u914d\u7f6e\u53c2\u6570\u8fdb\u884c\u5176\u4ed6\u7684\u64cd\u4f5c\u3002<\/p>\n

        1.4 \u81ea\u5b9a\u4e49 Nebula Graph Source \u5e94\u7528\u5b9e\u8df5<\/h3>\n

        \u4f7f\u7528 Flink \u8bfb\u53d6 Nebula Graph \u56fe\u6570\u636e\u65f6\uff0c\u9700\u8981\u6784\u9020 NebulaSourceFunction \u548c NebulaOutputFormat\uff0c\u5e76\u901a\u8fc7 Flink \u7684 addSource \u6216 createInput \u65b9\u6cd5\u6ce8\u518c\u6570\u636e\u6e90\u8fdb\u884c Nebula Graph \u6570\u636e\u8bfb\u53d6\u3002<\/p>\n

        \u6784\u9020 NebulaSourceFunction \u548c NebulaOutputFormat \u65f6\u9700\u8981\u8fdb\u884c\u5ba2\u6237\u7aef\u53c2\u6570\u7684\u914d\u7f6e\u548c\u6267\u884c\u53c2\u6570\u7684\u914d\u7f6e\uff0c\u8bf4\u660e\u5982\u4e0b\uff1a<\/p>\n

        \u914d\u7f6e\u9879\u8bf4\u660e\uff1a<\/p>\n

          \n
        • NebulaClientOptions\n
            \n
          • \u914d\u7f6e address\uff0cNebulaSource \u9700\u8981\u914d\u7f6e Nebula Graph Metad \u670d\u52a1\u7684\u5730\u5740\u3002<\/li>\n
          • \u914d\u7f6e username<\/li>\n
          • \u914d\u7f6e password<\/li>\n<\/ul>\n<\/li>\n
          • VertexExecutionOptions\n
              \n
            • \u914d\u7f6e GraphSpace<\/li>\n
            • \u914d\u7f6e\u8981\u8bfb\u53d6\u7684 tag<\/li>\n
            • \u914d\u7f6e\u8981\u8bfb\u53d6\u7684\u5b57\u6bb5\u96c6<\/li>\n
            • \u914d\u7f6e\u662f\u5426\u8bfb\u53d6\u6240\u6709\u5b57\u6bb5\uff0c\u9ed8\u8ba4\u4e3a false\uff0c \u82e5\u914d\u7f6e\u4e3a true \u5219\u5b57\u6bb5\u96c6\u914d\u7f6e\u65e0\u6548<\/li>\n
            • \u914d\u7f6e\u6bcf\u6b21\u8bfb\u53d6\u7684\u6570\u636e\u91cf limit\uff0c\u9ed8\u8ba4 2000<\/li>\n<\/ul>\n<\/li>\n
            • EdgeExecutionOptions\n
                \n
              • \u914d\u7f6e GraphSpace<\/li>\n
              • \u914d\u7f6e\u8981\u8bfb\u53d6\u7684 edge<\/li>\n
              • \u914d\u7f6e\u8981\u8bfb\u53d6\u7684\u5b57\u6bb5\u96c6<\/li>\n
              • \u914d\u7f6e\u662f\u5426\u8bfb\u53d6\u6240\u6709\u5b57\u6bb5\uff0c\u9ed8\u8ba4\u4e3a false\uff0c \u82e5\u914d\u7f6e\u4e3a true \u5219\u5b57\u6bb5\u96c6\u914d\u7f6e\u65e0\u6548<\/li>\n
              • \u914d\u7f6e\u6bcf\u6b21\u8bfb\u53d6\u7684\u6570\u636e\u91cf limit\uff0c\u9ed8\u8ba4 2000<\/li>\n<\/ul>\n<\/li>\n<\/ul>\n
                \/\/ \u6784\u9020 Nebula Graph \u5ba2\u6237\u7aef\u8fde\u63a5\u9700\u8981\u7684\u53c2\u6570 NebulaClientOptions nebulaClientOptions = new NebulaClientOptions                 .NebulaClientOptionsBuilder()                 .setAddress(\"127.0.0.1:45500\")                 .build(); \/\/ \u521b\u5efa connectionProvider NebulaConnectionProvider metaConnectionProvider = new NebulaMetaConnectionProvider(nebulaClientOptions);  \/\/ \u6784\u9020 Nebula Graph \u6570\u636e\u8bfb\u53d6\u9700\u8981\u7684\u53c2\u6570 List<String> cols = Arrays.asList(\"name\", \"age\"); VertexExecutionOptions sourceExecutionOptions = new VertexExecutionOptions.ExecutionOptionBuilder()                 .setGraphSpace(\"flinkSource\")                 .setTag(tag)                 .setFields(cols)                 .setLimit(100)                 .builder();  \/\/ \u6784\u9020 NebulaInputFormat NebulaInputFormat inputFormat = new NebulaInputFormat(metaConnectionProvider)                 .setExecutionOptions(sourceExecutionOptions);  \/\/ \u65b9\u5f0f 1 \u4f7f\u7528 createInput \u65b9\u5f0f\u6ce8\u518c Nebula Graph \u6570\u636e\u6e90 DataSource<Row> dataSource1 = ExecutionEnvironment.getExecutionEnvironment()          .createInput(inputFormat);   \/\/ \u65b9\u5f0f 2 \u4f7f\u7528 addSource \u65b9\u5f0f\u6ce8\u518c Nebula Graph \u6570\u636e\u6e90 NebulaSourceFunction sourceFunction = new NebulaSourceFunction(metaConnectionProvider)                 .setExecutionOptions(sourceExecutionOptions);  DataStreamSource<Row> dataSource2 = StreamExecutionEnvironment.getExecutionEnvironment()          .addSource(sourceFunction); <\/code><\/pre>\n
                Nebula Source Demo \u7f16\u5199\u5b8c\u6210\u540e\u53ef\u4ee5\u6253\u5305\u63d0\u4ea4\u5230 Flink \u96c6\u7fa4\u6267\u884c\u3002<\/p>\n
                \u793a\u4f8b\u7a0b\u5e8f\u8bfb\u53d6 Nebula Graph \u7684\u70b9\u6570\u636e\u5e76\u6253\u5370\uff0c\u8be5\u4f5c\u4e1a\u4ee5 Nebula Graph \u4f5c\u4e3a Source\uff0c\u4ee5 print \u4f5c\u4e3a Sink\uff0c\u6267\u884c\u7ed3\u679c\u5982\u4e0b\uff1a<\/p>\n
                \"Nebula<\/p>\n
                Source sent \u6570\u636e\u4e3a 59,671,064 \u6761\uff0cSink received \u6570\u636e\u4e3a 59,671,064 \u6761\u3002<\/p>\n
                \u4e8c\u3001Connector Sink<\/h2>\n
                Nebula Flink Connector \u4e2d\u7684 Sink \u5373 Nebula Graph \u56fe\u6570\u636e\u5e93\u3002Flink \u63d0\u4f9b\u4e86\u4e30\u5bcc\u7684 Connector \u7ec4\u4ef6\u5141\u8bb8\u7528\u6237\u81ea\u5b9a\u4e49\u6570\u636e\u6c60\u6765\u63a5\u6536 Flink \u6240\u5904\u7406\u7684\u6570\u636e\u6d41\u3002<\/p>\n
                2.1 Sink \u7b80\u4ecb<\/h3>\n
                Sink \u662f Flink \u5904\u7406\u5b8c Source \u540e\u6570\u636e\u7684\u8f93\u51fa\uff0c\u4e3b\u8981\u8d1f\u8d23\u5b9e\u65f6\u8ba1\u7b97\u7ed3\u679c\u7684\u8f93\u51fa\u548c\u6301\u4e45\u5316\u3002\u6bd4\u5982\uff1a\u5c06\u6570\u636e\u6d41\u5199\u5165\u6807\u51c6\u8f93\u51fa\u3001\u5199\u5165\u6587\u4ef6\u3001\u5199\u5165 Sockets \u3001\u5199\u5165\u5916\u90e8\u7cfb\u7edf\u7b49\u3002<\/p>\n
                Flink \u7684 Sink \u80fd\u529b\u4e3b\u8981\u662f\u901a\u8fc7\u8c03\u7528\u6570\u636e\u6d41\u7684 write \u76f8\u5173 API \u548c DataStream.addSink \u4e24\u79cd\u65b9\u5f0f\u6765\u5b9e\u73b0\u6570\u636e\u6d41\u7684\u5916\u90e8\u5b58\u50a8\u3002<\/p>\n
                \u7c7b\u4f3c\u4e8e Flink Connector \u7684 Source\uff0cSink \u4e5f\u5141\u8bb8\u7528\u6237\u81ea\u5b9a\u4e49\u6765\u652f\u6301\u4e30\u5bcc\u7684\u5916\u90e8\u6570\u636e\u7cfb\u7edf\u4f5c\u4e3a Flink \u7684\u6570\u636e\u6c60\u3002<\/p>\n
                Flink Sink \u7684\u4f7f\u7528\u65b9\u5f0f\u5982\u4e0b\uff1a<\/p>\n
                \"Nebula<\/p>\n
                \u672c\u7ae0\u4e3b\u8981\u4ecb\u7ecd\u5982\u4f55\u901a\u8fc7\u81ea\u5b9a\u4e49 Sink \u7684\u65b9\u5f0f\u5b9e\u73b0 Nebula Graph Sink \u3002<\/p>\n
                2.2 \u81ea\u5b9a\u4e49 Sink<\/h3>\n
                \u5728 Flink \u4e2d\u53ef\u4ee5\u4f7f\u7528 DataStream.addSink<\/code> \u548c DataStream.writeUsingOutputFormat<\/code> \u7684\u65b9\u5f0f\u5c06 Flink \u6570\u636e\u6d41\u5199\u5165\u5916\u90e8\u81ea\u5b9a\u4e49\u6570\u636e\u6c60\u3002<\/p>\n
                Flink \u5df2\u7ecf\u63d0\u4f9b\u4e86\u82e5\u5e72\u5b9e\u73b0\u597d\u4e86\u7684 Sink Functions<\/code> \uff0c\u4e5f\u53ef\u4ee5\u901a\u8fc7\u5b9e\u73b0 SinkFunction<\/code> \u4ee5\u53ca\u7ee7\u627f RichOutputFormat<\/code> \u6765\u5b9e\u73b0\u81ea\u5b9a\u4e49\u7684 Sink \u3002<\/p>\n
                2.3 \u81ea\u5b9a\u4e49 Nebula Graph Sink \u5b9e\u73b0\u539f\u7406<\/h3>\n
                Nebula Flink Connector \u4e2d\u5b9e\u73b0\u4e86\u81ea\u5b9a\u4e49\u7684 NebulaSinkFunction\uff0c\u5f00\u53d1\u8005\u901a\u8fc7\u8c03\u7528 DataSource.addSink \u65b9\u6cd5\u5e76\u5c06 NebulaSinkFunction \u5bf9\u8c61\u4f5c\u4e3a\u53c2\u6570\u4f20\u5165\u5373\u53ef\u5b9e\u73b0\u5c06 Flink \u6570\u636e\u6d41\u5199\u5165 Nebula Graph \u3002<\/p>\n
                Nebula Flink Connector \u4f7f\u7528\u7684\u662f Flink \u7684 1.11-SNAPSHOT \u7248\u672c\uff0c\u8be5\u7248\u672c\u4e2d\u5df2\u7ecf\u5e9f\u5f03\u4e86\u4f7f\u7528 writeUsingOutputFormat \u65b9\u6cd5\u6765\u5b9a\u4e49\u8f93\u51fa\u7aef\u7684\u63a5\u53e3\u3002<\/p>\n
                \u6e90\u7801\u5982\u4e0b\uff0c\u6240\u4ee5\u8bf7\u6ce8\u610f\u5728\u4f7f\u7528\u81ea\u5b9a\u4e49 Nebula Graph Sink \u65f6\u8bf7\u91c7\u7528 DataStream.addSink \u7684\u65b9\u5f0f\u3002<\/p>\n
                    \/** @deprecated *\/     @Deprecated     @PublicEvolving     public DataStreamSink<T> writeUsingOutputFormat(OutputFormat<T> format) {         return this.addSink(new OutputFormatSinkFunction(format));     } <\/code><\/pre>\n
                Nebula Graph Sink \u5b9e\u73b0\u7c7b\u56fe\u5982\u4e0b\uff1a<\/p>\n
                \"Nebula<\/p>\n
                \u5176\u4e2d\u6700\u91cd\u8981\u7684\u4e24\u4e2a\u7c7b\u662f NebulaSinkFunction \u548c NebulaBatchOutputFormat \u3002<\/p>\n
                NebulaSinkFunction \u7ee7\u627f\u81ea AbstractRichFunction \u5e76\u5b9e\u73b0\u4e86\u4ee5\u4e0b\u65b9\u6cd5\uff1a<\/p>\n
                  \n
                • open \u8c03\u7528 NebulaBatchOutputFormat \u7684 open \u65b9\u6cd5\uff0c\u8fdb\u884c\u8d44\u6e90\u51c6\u5907\u3002<\/li>\n
                • close \u8c03\u7528 NebulaBatchOutputFormat \u7684 close \u65b9\u6cd5\uff0c\u8fdb\u884c\u8d44\u6e90\u91ca\u653e\u3002<\/li>\n
                • invoke \u662f Sink \u4e2d\u7684\u6838\u5fc3\u65b9\u6cd5\uff0c \u8c03\u7528 NebulaBatchOutputFormat \u4e2d\u7684 write \u65b9\u6cd5\u8fdb\u884c\u6570\u636e\u5199\u5165\u3002<\/li>\n
                • flush \u8c03\u7528 NebulaBatchOutputFormat \u7684 flush \u65b9\u6cd5\u8fdb\u884c\u6570\u636e\u7684\u63d0\u4ea4\u3002<\/li>\n<\/ul>\n
                  NebulaBatchOutputFormat \u7ee7\u627f\u81ea AbstractNebulaOutPutFormat\uff0cAbstractNebulaOutPutFormat \u7ee7\u627f\u81ea RichOutputFormat\uff0c\u4e3b\u8981\u5b9e\u73b0\u7684\u65b9\u6cd5\u6709\uff1a<\/p>\n
                    \n
                  • open \u51c6\u5907\u56fe\u6570\u636e\u5e93 Nebula Graph \u7684 Graphd \u670d\u52a1\u7684\u8fde\u63a5\uff0c\u5e76\u521d\u59cb\u5316\u6570\u636e\u5199\u5165\u6267\u884c\u5668 nebulaBatchExecutor<\/li>\n
                  • close \u63d0\u4ea4\u6700\u540e\u6279\u6b21\u6570\u636e\uff0c\u7b49\u5f85\u6700\u540e\u63d0\u4ea4\u7684\u56de\u8c03\u7ed3\u679c\u5e76\u5173\u95ed\u670d\u52a1\u8fde\u63a5\u7b49\u8d44\u6e90\u3002<\/li>\n
                  • writeRecord \u6838\u5fc3\u65b9\u6cd5\uff0c\u5c06\u6570\u636e\u5199\u5165 nebulaBufferedRow \u4e2d\uff0c\u5e76\u5728\u8fbe\u5230\u914d\u7f6e\u7684\u6279\u91cf\u5199\u5165 Nebula Graph \u4e0a\u9650\u65f6\u63d0\u4ea4\u5199\u5165\u3002Nebula Graph Sink \u7684\u5199\u5165\u64cd\u4f5c\u662f\u5f02\u6b65\u7684\uff0c\u6240\u4ee5\u9700\u8981\u6267\u884c\u56de\u8c03\u6765\u83b7\u53d6\u6267\u884c\u7ed3\u679c\u3002<\/li>\n
                  • flush \u5f53 bufferRow \u5b58\u5728\u6570\u636e\u65f6\uff0c\u5c06\u6570\u636e\u63d0\u4ea4\u5230 Nebula Graph \u4e2d\u3002<\/li>\n<\/ul>\n
                    \u5728 AbstractNebulaOutputFormat \u4e2d\u8c03\u7528\u4e86 NebulaBatchExecutor \u8fdb\u884c\u6570\u636e\u7684\u6279\u91cf\u7ba1\u7406\u548c\u6279\u91cf\u63d0\u4ea4\uff0c\u5e76\u901a\u8fc7\u5b9a\u4e49\u56de\u8c03\u51fd\u6570\u63a5\u6536\u6279\u91cf\u63d0\u4ea4\u7684\u7ed3\u679c\uff0c\u4ee3\u7801\u5982\u4e0b\uff1a<\/p>\n
                        \/**      * write one record to buffer      *\/     @Override     public final synchronized void writeRecord(T row) throws IOException {         nebulaBatchExecutor.addToBatch(row);          if (numPendingRow.incrementAndGet() >= executionOptions.getBatch()) {             commit();         }     }      \/**      * put record into buffer      *      * @param record represent vertex or edge      *\/     void addToBatch(T record) {         boolean isVertex = executionOptions.getDataType().isVertex();          NebulaOutputFormatConverter converter;         if (isVertex) {             converter = new NebulaRowVertexOutputFormatConverter((VertexExecutionOptions) executionOptions);         } else {             converter = new NebulaRowEdgeOutputFormatConverter((EdgeExecutionOptions) executionOptions);         }         String value = converter.createValue(record, executionOptions.getPolicy());         if (value == null) {             return;         }         nebulaBufferedRow.putRow(value);     }      \/**      * commit batch insert statements      *\/     private synchronized void commit() throws IOException {         graphClient.switchSpace(executionOptions.getGraphSpace());         future = nebulaBatchExecutor.executeBatch(graphClient);         \/\/ clear waiting rows         numPendingRow.compareAndSet(executionOptions.getBatch(),0);     }      \/**      * execute the insert statement      *      * @param client Asynchronous graph client      *\/     ListenableFuture executeBatch(AsyncGraphClientImpl client) {         String propNames = String.join(NebulaConstant.COMMA, executionOptions.getFields());         String values = String.join(NebulaConstant.COMMA, nebulaBufferedRow.getRows());         \/\/ construct insert statement         String exec = String.format(NebulaConstant.BATCH_INSERT_TEMPLATE, executionOptions.getDataType(), executionOptions.getLabel(), propNames, values);         \/\/ execute insert statement         ListenableFuture<Optional<Integer>> execResult = client.execute(exec);         \/\/ define callback function         Futures.addCallback(execResult, new FutureCallback<Optional<Integer>>() {             @Override             public void onSuccess(Optional<Integer> integerOptional) {                 if (integerOptional.isPresent()) {                     if (integerOptional.get() == ErrorCode.SUCCEEDED) {                         LOG.info(\"batch insert Succeed\");                     } else {                         LOG.error(String.format(\"batch insert Error: %d\",                                 integerOptional.get()));                     }                 } else {                     LOG.error(\"batch insert Error\");                 }             }              @Override             public void onFailure(Throwable throwable) {                 LOG.error(\"batch insert Error\");             }         });         nebulaBufferedRow.clean();         return execResult;     } <\/code><\/pre>\n
                    \u7531\u4e8e Nebula Graph Sink \u7684\u5199\u5165\u662f\u6279\u91cf\u3001\u5f02\u6b65\u7684\uff0c\u6240\u4ee5\u5728\u6700\u540e\u4e1a\u52a1\u7ed3\u675f close \u8d44\u6e90\u4e4b\u524d\u9700\u8981\u5c06\u7f13\u5b58\u4e2d\u7684\u6279\u91cf\u6570\u636e\u63d0\u4ea4\u4e14\u7b49\u5f85\u5199\u5165\u64cd\u4f5c\u7684\u5b8c\u6210\uff0c\u4ee5\u9632\u5728\u5199\u5165\u63d0\u4ea4\u4e4b\u524d\u63d0\u524d\u628a Nebula Graph Client \u5173\u95ed\uff0c\u4ee3\u7801\u5982\u4e0b\uff1a<\/p>\n
                        \/**      * commit the batch write operator before release connection      *\/     @Override     public  final synchronized void close() throws IOException {         if(numPendingRow.get() > 0){             commit();         }         while(!future.isDone()){             try {                 Thread.sleep(10);             } catch (InterruptedException e) {                 LOG.error(\"sleep interrupted, \", e);             }         }          super.close();     }  <\/code><\/pre>\n
                    2.4 \u81ea\u5b9a\u4e49 Nebula Graph Sink \u5e94\u7528\u5b9e\u8df5<\/h3>\n
                    Flink \u5c06\u5904\u7406\u5b8c\u6210\u7684\u6570\u636e Sink \u5230 Nebula Graph \u65f6\uff0c\u9700\u8981\u5c06 Flink \u6570\u636e\u6d41\u8fdb\u884c map \u8f6c\u6362\u6210 Nebula Graph Sink \u53ef\u63a5\u6536\u7684\u6570\u636e\u683c\u5f0f\u3002\u81ea\u5b9a\u4e49 Nebula Graph Sink \u7684\u4f7f\u7528\u65b9\u5f0f\u662f\u901a\u8fc7 addSink \u5f62\u5f0f\uff0c\u5c06 NebulaSinkFunction \u4f5c\u4e3a\u53c2\u6570\u4f20\u7ed9 addSink \u65b9\u6cd5\u6765\u5b9e\u73b0 Flink \u6570\u636e\u6d41\u7684\u5199\u5165\u3002<\/p>\n
                      \n
                    • NebulaClientOptions\n
                        \n
                      • \u914d\u7f6e address\uff0cNebulaSource \u9700\u8981\u914d\u7f6e Nebula Graph Graphd \u670d\u52a1\u7684\u5730\u5740\u3002<\/li>\n
                      • \u914d\u7f6e username<\/li>\n
                      • \u914d\u7f6e password<\/li>\n<\/ul>\n<\/li>\n
                      • VertexExecutionOptions\n
                          \n
                        • \u914d\u7f6e GraphSpace<\/li>\n
                        • \u914d\u7f6e\u8981\u5199\u5165\u7684 tag<\/li>\n
                        • \u914d\u7f6e\u8981\u5199\u5165\u7684\u5b57\u6bb5\u96c6<\/li>\n
                        • \u914d\u7f6e\u5199\u5165\u7684\u70b9 ID \u6240\u5728 Flink \u6570\u636e\u6d41 Row \u4e2d\u7684\u7d22\u5f15<\/li>\n
                        • \u914d\u7f6e\u6279\u91cf\u5199\u5165 Nebula Graph \u7684\u6570\u91cf\uff0c\u9ed8\u8ba4 2000<\/li>\n<\/ul>\n<\/li>\n
                        • EdgeExecutionOptions\n
                            \n
                          • \u914d\u7f6e GraphSpace<\/li>\n
                          • \u914d\u7f6e\u8981\u5199\u5165\u7684 edge<\/li>\n
                          • \u914d\u7f6e\u8981\u5199\u5165\u7684\u5b57\u6bb5\u96c6<\/li>\n
                          • \u914d\u7f6e\u5199\u5165\u7684\u8fb9 src-id \u6240\u5728 Flink \u6570\u636e\u6d41 Row \u4e2d\u7684\u7d22\u5f15<\/li>\n
                          • \u914d\u7f6e\u5199\u5165\u7684\u8fb9 dst-id \u6240\u5728 Flink \u6570\u636e\u6d41 Row \u4e2d\u7684\u7d22\u5f15<\/li>\n
                          • \u914d\u7f6e\u5199\u5165\u7684\u8fb9 rank \u6240\u5728 Flink \u6570\u636e\u6d41 Row \u4e2d\u7684\u7d22\u5f15\uff0c\u4e0d\u914d\u5219\u65e0 rank<\/li>\n
                          • \u914d\u7f6e\u6279\u91cf\u5199\u5165 Nebula Graph \u7684\u6570\u91cf\uff0c\u9ed8\u8ba4 2000<\/li>\n<\/ul>\n<\/li>\n<\/ul>\n
                            \/\/\/ \u6784\u9020 Nebula Graphd \u5ba2\u6237\u7aef\u8fde\u63a5\u9700\u8981\u7684\u53c2\u6570 NebulaClientOptions nebulaClientOptions = new NebulaClientOptions                 .NebulaClientOptionsBuilder()                 .setAddress(\"127.0.0.1:3699\")                 .build(); NebulaConnectionProvider graphConnectionProvider = new NebulaGraphConnectionProvider(nebulaClientOptions);   \/\/ \u6784\u9020 Nebula Graph \u5199\u5165\u64cd\u4f5c\u53c2\u6570 List<String> cols = Arrays.asList(\"name\", \"age\") ExecutionOptions sinkExecutionOptions = new VertexExecutionOptions.ExecutionOptionBuilder()                 .setGraphSpace(\"flinkSink\")                 .setTag(tag)                 .setFields(cols)                 .setIdIndex(0)                 .setBatch(20)                 .builder();    \/\/ \u5199\u5165 Nebula Graph dataSource.addSink(nebulaSinkFunction); <\/code><\/pre>\n
                            Nebula Graph Sink \u7684 Demo \u7a0b\u5e8f\u4ee5 Nebula Graph \u7684 space:flinkSource \u4f5c\u4e3a Source \u8bfb\u53d6\u6570\u636e\uff0c\u8fdb\u884c map \u7c7b\u578b\u8f6c\u6362\u540e Sink \u5165 Nebula Graph \u7684 space:flinkSink\uff0c\u5bf9\u5e94\u7684\u5e94\u7528\u573a\u666f\u4e3a\u5c06 Nebula Graph \u4e2d\u4e00\u4e2a space \u7684\u6570\u636e\u6d41\u5165\u53e6\u4e00\u4e2a space \u4e2d\u3002<\/p>\n
                            \u4e09\u3001Catalog<\/h2>\n
                            Flink 1.11.0 \u4e4b\u524d\uff0c\u7528\u6237\u5982\u679c\u4f9d\u8d56 Flink \u7684 Source\/Sink \u8bfb\u5199\u5916\u90e8\u6570\u636e\u6e90\u65f6\uff0c\u5fc5\u987b\u8981\u624b\u52a8\u8bfb\u53d6\u5bf9\u5e94\u6570\u636e\u7cfb\u7edf\u7684 Schema \u3002\u6bd4\u5982\uff0c\u8981\u8bfb\u5199 Nebula Graph\uff0c\u5219\u5fc5\u987b\u5148\u4fdd\u8bc1\u660e\u786e\u5730\u77e5\u6653\u5728 Nebula Graph \u4e2d\u7684 Schema \u4fe1\u606f\u3002\u4f46\u662f\u8fd9\u6837\u4f1a\u6709\u4e00\u4e2a\u95ee\u9898\uff0c\u5f53 Nebula Graph \u4e2d\u7684 Schema \u53d1\u751f\u53d8\u5316\u65f6\uff0c\u4e5f\u9700\u8981\u624b\u52a8\u66f4\u65b0\u5bf9\u5e94\u7684 Flink \u4efb\u52a1\u4ee5\u4fdd\u6301\u7c7b\u578b\u5339\u914d\uff0c\u4efb\u4f55\u4e0d\u5339\u914d\u90fd\u4f1a\u9020\u6210\u8fd0\u884c\u65f6\u62a5\u9519\u4f7f\u4f5c\u4e1a\u5931\u8d25\u3002\u8fd9\u4e2a\u64cd\u4f5c\u5197\u4f59\u4e14\u7e41\u7410\uff0c\u4f53\u9a8c\u6781\u5dee\u3002<\/p>\n
                            1.11.0 \u7248\u672c\u540e\uff0c\u7528\u6237\u4f7f\u7528 Flink Connector \u65f6\u53ef\u4ee5\u81ea\u52a8\u83b7\u53d6\u8868\u7684 Schema \u3002\u53ef\u4ee5\u5728\u4e0d\u4e86\u89e3\u5916\u90e8\u7cfb\u7edf\u6570\u636e Schema \u7684\u60c5\u51b5\u4e0b\u8fdb\u884c\u6570\u636e\u5339\u914d\u3002<\/p>\n
                            \u76ee\u524d Nebula Flink Connector \u4e2d\u5df2\u652f\u6301\u6570\u636e\u7684\u8bfb\u5199\uff0c\u8981\u5b9e\u73b0 Schema \u7684\u5339\u914d\u5219\u9700\u8981\u4e3a Flink Connector \u5b9e\u73b0 Catalog \u7684\u7ba1\u7406\u3002\u4f46\u4e3a\u4e86\u786e\u4fdd Nebula Graph \u4e2d\u6570\u636e\u7684\u5b89\u5168\u6027\uff0cNebula Flink Connector \u53ea\u652f\u6301 Catalog \u7684\u8bfb\u64cd\u4f5c\uff0c\u4e0d\u5141\u8bb8\u8fdb\u884c Catalog \u7684\u4fee\u6539\u548c\u5199\u5165\u3002<\/p>\n
                            \u8bbf\u95ee Nebula Graph \u6307\u5b9a\u7c7b\u578b\u7684\u6570\u636e\u65f6\uff0c\u5b8c\u6574\u8def\u5f84\u5e94\u8be5\u662f\u4ee5\u4e0b\u683c\u5f0f\uff1a<graphSpace>.<VERTEX.tag><\/code> \u6216\u8005 <graphSpace>.<EDGE.edge><\/code><\/p>\n
                            \u5177\u4f53\u4f7f\u7528\u65b9\u5f0f\u5982\u4e0b\uff1a<\/p>\n
                            String catalogName  = \"testCatalog\"; String defaultSpace = \"flinkSink\"; String username     = \"root\"; String password     = \"nebula\"; String address      = \"127.0.0.1:45500\"; String table        = \"VERTEX.player\"  \/\/ define Nebula catalog Catalog catalog = NebulaCatalogUtils.createNebulaCatalog(catalogName,defaultSpace, address, username, password); \/\/ define Flink table environment StreamExecutionEnvironment bsEnv = StreamExecutionEnvironment.getExecutionEnvironment(); tEnv = StreamTableEnvironment.create(bsEnv); \/\/ register customed nebula catalog tEnv.registerCatalog(catalogName, catalog); \/\/ use customed nebula catalog tEnv.useCatalog(catalogName);  \/\/ show graph spaces of Nebula Graph String[] spaces = tEnv.listDatabases();  \/\/ show tags and edges of Nebula Graph tEnv.useDatabase(defaultSpace); String[] tables = tEnv.listTables();  \/\/ check tag player exist in defaultSpace ObjectPath path = new ObjectPath(defaultSpace, table); assert catalog.tableExists(path) == true      \/\/ get nebula tag schema CatalogBaseTable table = catalog.getTable(new ObjectPath(defaultSpace, table)); table.getSchema(); <\/code><\/pre>\n
                            Nebula Flink Connector \u652f\u6301\u7684\u5176\u4ed6 Catalog \u63a5\u53e3\u8bf7\u67e5\u770b GitHub \u4ee3\u7801 NebulaCatalog.java \u3002<\/p>\n
                            \u56db\u3001Exactly-once<\/h2>\n
                            Flink Connector \u7684 Exactly-once \u662f\u6307 Flink \u501f\u52a9\u4e8e checkpoint \u673a\u5236\u4fdd\u8bc1\u6bcf\u4e2a\u8f93\u5165\u4e8b\u4ef6\u53ea\u5bf9\u6700\u7ec8\u7ed3\u679c\u5f71\u54cd\u4e00\u6b21\uff0c\u5728\u6570\u636e\u5904\u7406\u8fc7\u7a0b\u4e2d\u5373\u4f7f\u51fa\u73b0\u6545\u969c\uff0c\u4e5f\u4e0d\u4f1a\u5b58\u5728\u6570\u636e\u91cd\u590d\u548c\u4e22\u5931\u7684\u60c5\u51b5\u3002<\/p>\n
                            \u4e3a\u4e86\u63d0\u4f9b\u7aef\u5230\u7aef\u7684 Exactly-once \u8bed\u4e49\uff0cFlink \u7684\u5916\u90e8\u6570\u636e\u7cfb\u7edf\u4e5f\u5fc5\u987b\u63d0\u4f9b\u63d0\u4ea4\u6216\u56de\u6eda\u7684\u65b9\u6cd5\uff0c\u7136\u540e\u901a\u8fc7 Flink \u7684 checkpoint \u673a\u5236\u534f\u8c03\u3002Flink \u63d0\u4f9b\u4e86\u5b9e\u73b0\u7aef\u5230\u7aef\u7684 Exactly-once \u7684\u62bd\u8c61\uff0c\u5373\u5b9e\u73b0\u4e8c\u9636\u6bb5\u63d0\u4ea4\u7684\u62bd\u8c61\u7c7b TwoPhaseCommitSinkFunction \u3002<\/p>\n
                            \u60f3\u4e3a\u6570\u636e\u8f93\u51fa\u7aef\u5b9e\u73b0 Exactly-once\uff0c\u5219\u9700\u8981\u5b9e\u73b0\u56db\u4e2a\u51fd\u6570\uff1a<\/p>\n
                              \n
                            • beginTransaction \u5728\u4e8b\u52a1\u5f00\u59cb\u524d\uff0c\u5728\u76ee\u6807\u6587\u4ef6\u7cfb\u7edf\u7684\u4e34\u65f6\u76ee\u5f55\u521b\u5efa\u4e00\u4e2a\u4e34\u65f6\u6587\u4ef6\uff0c\u968f\u540e\u53ef\u4ee5\u5728\u6570\u636e\u5904\u7406\u65f6\u5c06\u6570\u636e\u5199\u5165\u6b64\u6587\u4ef6\u3002<\/li>\n
                            • preCommit \u5728\u9884\u63d0\u4ea4\u9636\u6bb5\uff0c\u5173\u95ed\u6587\u4ef6\u4e0d\u518d\u5199\u5165\u3002\u4e3a\u4e0b\u4e00\u4e2a checkpoint \u7684\u4efb\u4f55\u540e\u7eed\u6587\u4ef6\u5199\u5165\u542f\u52a8\u4e00\u4e2a\u65b0\u4e8b\u52a1\u3002<\/li>\n
                            • commit \u5728\u63d0\u4ea4\u9636\u6bb5\uff0c\u5c06\u9884\u63d0\u4ea4\u9636\u6bb5\u7684\u6587\u4ef6\u539f\u5b50\u5730\u79fb\u52a8\u5230\u771f\u6b63\u7684\u76ee\u6807\u76ee\u5f55\u3002\u4e8c\u9636\u6bb5\u63d0\u4ea4\u8fc7\u7a0b\u4f1a\u589e\u52a0\u8f93\u51fa\u6570\u636e\u53ef\u89c1\u6027\u7684\u5ef6\u8fdf\u3002<\/li>\n
                            • abort \u5728\u7ec8\u6b62\u9636\u6bb5\uff0c\u5220\u9664\u4e34\u65f6\u6587\u4ef6\u3002<\/li>\n<\/ul>\n
                              \u6839\u636e\u4e0a\u8ff0\u51fd\u6570\u53ef\u770b\u51fa\uff0cFlink \u7684\u4e8c\u9636\u6bb5\u63d0\u4ea4\u5bf9\u5916\u90e8\u6570\u636e\u6e90\u6709\u8981\u6c42\uff0c\u5373 Source \u6570\u636e\u6e90\u5fc5\u987b\u5177\u5907\u91cd\u53d1\u529f\u80fd\uff0cSink \u6570\u636e\u6c60\u5fc5\u987b\u652f\u6301\u4e8b\u52a1\u63d0\u4ea4\u548c\u5e42\u7b49\u5199\u3002<\/p>\n
                              Nebula Graph v1.1.0 \u867d\u7136\u4e0d\u652f\u6301\u4e8b\u52a1\uff0c\u4f46\u5176\u5199\u5165\u64cd\u4f5c\u662f\u5e42\u7b49\u7684\uff0c\u5373\u540c\u4e00\u6761\u6570\u636e\u7684\u591a\u6b21\u5199\u5165\u7ed3\u679c\u662f\u4e00\u81f4\u7684\u3002\u56e0\u6b64\u53ef\u4ee5\u901a\u8fc7 checkpoint \u673a\u5236\u5b9e\u73b0 Nebula Flink Connector \u7684 At-least-Once \u673a\u5236\uff0c\u6839\u636e\u591a\u6b21\u5199\u5165\u7684\u5e42\u7b49\u6027\u53ef\u4ee5\u95f4\u63a5\u5b9e\u73b0 Sink \u7684 Exactly-once \u3002<\/p>\n
                              \u8981\u4f7f\u7528 Nebula Graph Sink \u7684\u5bb9\u9519\u6027\uff0c\u8bf7\u786e\u4fdd\u5728 Flink \u7684\u6267\u884c\u73af\u5883\u4e2d\u5f00\u542f\u4e86 checkpoint \u914d\u7f6e\uff1a<\/p>\n
                              StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.enableCheckpointing(10000) \/\/ checkpoint every 10000 msecs    .getCheckpointConfig()    .setCheckpointingMode(CheckpointingMode.AT_LEAST_ONCE); <\/code><\/pre>\n
                              Reference<\/h2>\n
                                \n
                              • Nebula Source Demo [testNebulaSource]\uff1ahttps:\/\/github.com\/vesoft-inc\/nebula-java\/blob\/master\/examples\/src\/main\/java\/org\/apache\/flink\/FlinkDemo.java<\/li>\n
                              • Nebula Sink Demo [testSourceSink]\uff1ahttps:\/\/github.com\/vesoft-inc\/nebula-java\/blob\/master\/examples\/src\/main\/java\/org\/apache\/flink\/FlinkDemo.java<\/li>\n
                              • Apache Flink \u6e90\u7801\uff1ahttps:\/\/github.com\/apache\/flink<\/li>\n
                              • ApacheFlink \u96f6\u57fa\u7840\u5165\u95e8\uff1ahttps:\/\/www.infoq.cn\/theme\/28<\/li>\n
                              • Flink \u6587\u6863\uff1ahttps:\/\/flink.apache.org\/flink-architecture.html<\/li>\n
                              • Flink \u5b9e\u8df5\u6587\u6863\uff1ahttps:\/\/ci.apache.org\/projects\/flink\/flink-docs-release-1.12\/<\/li>\n
                              • flink-connector-jdbc \u6e90\u7801\uff1ahttps:\/\/github.com\/apache\/flink\/tree\/master\/flink-connectors\/flink-connector-jdbc<\/li>\n
                              • Flink JDBC Catalog \u8be6\u89e3\uff1ahttps:\/\/cloud.tencent.com\/developer\/article\/1697913<\/li>\n<\/ul>\n
                                \u559c\u6b22\u8fd9\u7bc7\u6587\u7ae0\uff1f\u6765\u6765\u6765\uff0c\u7ed9\u6211\u4eec\u7684 GitHub \u70b9\u4e2a star \u8868\u9f13\u52b1\u5566~~       <\/p><\/div>\n
                                 \u5927\u4f6c\u6709\u8a71\u8aaa<\/b> (0<\/span>)        <\/div>\n
                                 <\/div>\n<\/p><\/div>\n<\/p><\/div>\n
                                  \n
                                • \n","protected":false},"excerpt":{"rendered":"
                                  Nebula Flink Conn…<\/p>\n","protected":false},"author":1,"featured_media":0,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":[],"categories":[],"tags":[],"_links":{"self":[{"href":"http:\/\/4563.org\/index.php?rest_route=\/wp\/v2\/posts\/217480"}],"collection":[{"href":"http:\/\/4563.org\/index.php?rest_route=\/wp\/v2\/posts"}],"about":[{"href":"http:\/\/4563.org\/index.php?rest_route=\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"http:\/\/4563.org\/index.php?rest_route=\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"http:\/\/4563.org\/index.php?rest_route=%2Fwp%2Fv2%2Fcomments&post=217480"}],"version-history":[{"count":0,"href":"http:\/\/4563.org\/index.php?rest_route=\/wp\/v2\/posts\/217480\/revisions"}],"wp:attachment":[{"href":"http:\/\/4563.org\/index.php?rest_route=%2Fwp%2Fv2%2Fmedia&parent=217480"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"http:\/\/4563.org\/index.php?rest_route=%2Fwp%2Fv2%2Fcategories&post=217480"},{"taxonomy":"post_tag","embeddable":true,"href":"http:\/\/4563.org\/index.php?rest_route=%2Fwp%2Fv2%2Ftags&post=217480"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}