{"id":1437,"date":"2020-05-28T01:37:38","date_gmt":"2020-05-27T17:37:38","guid":{"rendered":"http:\/\/39.97.231.10\/?p=1437"},"modified":"2020-05-28T01:37:38","modified_gmt":"2020-05-27T17:37:38","slug":"hudi-%e6%95%b0%e6%8d%ae%e9%9b%86%e5%86%99%e5%85%a5","status":"publish","type":"post","link":"http:\/\/www.weijing.co\/?p=1437","title":{"rendered":"Hudi \u6570\u636e\u96c6\u5199\u5165"},"content":{"rendered":"
<\/header>\n
\u00a0 \u00a0 \u00a0 \u00a0 \u4ecb\u7ecd\u4f7f\u7528DeltaStreamer\u5de5\u5177\u4ece\u5916\u90e8\u6e90\u751a\u81f3\u5176\u4ed6Hudi\u6570\u636e\u96c6\u6444\u53d6\u65b0\u66f4\u6539\u7684\u65b9\u6cd5\uff0c \u4ee5\u53ca\u901a\u8fc7\u4f7f\u7528Hudi\u6570\u636e\u6e90\u7684upserts\u52a0\u5feb\u5927\u578bSpark\u4f5c\u4e1a\u7684\u65b9\u6cd5\u3002 \u5bf9\u4e8e\u6b64\u7c7b\u6570\u636e\u96c6\uff0c\u6211\u4eec\u53ef\u4ee5\u4f7f\u7528\u5404\u79cd\u67e5\u8be2\u5f15\u64ce\u67e5\u8be2\u5b83\u4eec\u3002<\/p>\n

\u5199\u64cd\u4f5cPermalink<\/span><\/i><\/h2>\n

\u5728\u6b64\u4e4b\u524d\uff0c\u4e86\u89e3Hudi\u6570\u636e\u6e90\u53cadelta streamer\u5de5\u5177\u63d0\u4f9b\u7684\u4e09\u79cd\u4e0d\u540c\u7684\u5199\u64cd\u4f5c\u4ee5\u53ca\u5982\u4f55\u6700\u4f73\u5229\u7528\u5b83\u4eec\u53ef\u80fd\u4f1a\u6709\u6240\u5e2e\u52a9\u3002 \u8fd9\u4e9b\u64cd\u4f5c\u53ef\u4ee5\u5728\u9488\u5bf9\u6570\u636e\u96c6\u53d1\u51fa\u7684\u6bcf\u4e2a\u63d0\u4ea4\/\u589e\u91cf\u63d0\u4ea4\u4e2d\u8fdb\u884c\u9009\u62e9\/\u66f4\u6539\u3002<\/p>\n

    \n
  • UPSERT\uff08\u63d2\u5165\u66f4\u65b0\uff09<\/strong>\u00a0\uff1a\u8fd9\u662f\u9ed8\u8ba4\u64cd\u4f5c\uff0c\u5728\u8be5\u64cd\u4f5c\u4e2d\uff0c\u901a\u8fc7\u67e5\u627e\u7d22\u5f15\uff0c\u9996\u5148\u5c06\u8f93\u5165\u8bb0\u5f55\u6807\u8bb0\u4e3a\u63d2\u5165\u6216\u66f4\u65b0\u3002 \u5728\u8fd0\u884c\u542f\u53d1\u5f0f\u65b9\u6cd5\u4ee5\u786e\u5b9a\u5982\u4f55\u6700\u597d\u5730\u5c06\u8fd9\u4e9b\u8bb0\u5f55\u653e\u5230\u5b58\u50a8\u4e0a\uff0c\u5982\u4f18\u5316\u6587\u4ef6\u5927\u5c0f\u4e4b\u540e\uff0c\u8fd9\u4e9b\u8bb0\u5f55\u6700\u7ec8\u4f1a\u88ab\u5199\u5165\u3002 \u5bf9\u4e8e\u8bf8\u5982\u6570\u636e\u5e93\u66f4\u6539\u6355\u83b7\u4e4b\u7c7b\u7684\u7528\u4f8b\uff0c\u5efa\u8bae\u8be5\u64cd\u4f5c\uff0c\u56e0\u4e3a\u8f93\u5165\u51e0\u4e4e\u80af\u5b9a\u5305\u542b\u66f4\u65b0\u3002<\/li>\n
  • INSERT\uff08\u63d2\u5165\uff09<\/strong>\u00a0\uff1a\u5c31\u4f7f\u7528\u542f\u53d1\u5f0f\u65b9\u6cd5\u786e\u5b9a\u6587\u4ef6\u5927\u5c0f\u800c\u8a00\uff0c\u6b64\u64cd\u4f5c\u4e0e\u63d2\u5165\u66f4\u65b0\uff08UPSERT\uff09\u975e\u5e38\u76f8\u4f3c\uff0c\u4f46\u6b64\u64cd\u4f5c\u5b8c\u5168\u8df3\u8fc7\u4e86\u7d22\u5f15\u67e5\u627e\u6b65\u9aa4\u3002 \u56e0\u6b64\uff0c\u5bf9\u4e8e\u65e5\u5fd7\u91cd\u590d\u6570\u636e\u5220\u9664\u7b49\u7528\u4f8b\uff08\u7ed3\u5408\u4e0b\u9762\u63d0\u5230\u7684\u8fc7\u6ee4\u91cd\u590d\u9879\u7684\u9009\u9879\uff09\uff0c\u5b83\u53ef\u4ee5\u6bd4\u63d2\u5165\u66f4\u65b0\u5feb\u5f97\u591a\u3002 \u63d2\u5165\u4e5f\u9002\u7528\u4e8e\u8fd9\u79cd\u7528\u4f8b\uff0c\u8fd9\u79cd\u60c5\u51b5\u6570\u636e\u96c6\u53ef\u4ee5\u5141\u8bb8\u91cd\u590d\u9879\uff0c\u4f46\u53ea\u9700\u8981Hudi\u7684\u4e8b\u52a1\u5199\/\u589e\u91cf\u63d0\u53d6\/\u5b58\u50a8\u7ba1\u7406\u529f\u80fd\u3002<\/li>\n
  • BULK_INSERT\uff08\u6279\u63d2\u5165\uff09<\/strong>\u00a0\uff1a\u63d2\u5165\u66f4\u65b0\u548c\u63d2\u5165\u64cd\u4f5c\u90fd\u5c06\u8f93\u5165\u8bb0\u5f55\u4fdd\u5b58\u5728\u5185\u5b58\u4e2d\uff0c\u4ee5\u52a0\u5feb\u5b58\u50a8\u4f18\u5316\u542f\u53d1\u5f0f\u8ba1\u7b97\u7684\u901f\u5ea6\uff08\u4ee5\u53ca\u5176\u5b83\u672a\u63d0\u53ca\u7684\u65b9\u9762\uff09\u3002 \u6240\u4ee5\u5bf9Hudi\u6570\u636e\u96c6\u8fdb\u884c\u521d\u59cb\u52a0\u8f7d\/\u5f15\u5bfc\u65f6\u8fd9\u4e24\u79cd\u64cd\u4f5c\u4f1a\u5f88\u4f4e\u6548\u3002\u6279\u91cf\u63d2\u5165\u63d0\u4f9b\u4e0e\u63d2\u5165\u76f8\u540c\u7684\u8bed\u4e49\uff0c\u4f46\u540c\u65f6\u5b9e\u73b0\u4e86\u57fa\u4e8e\u6392\u5e8f\u7684\u6570\u636e\u5199\u5165\u7b97\u6cd5\uff0c \u8be5\u7b97\u6cd5\u53ef\u4ee5\u5f88\u597d\u5730\u6269\u5c55\u6570\u767eTB\u7684\u521d\u59cb\u8d1f\u8f7d\u3002\u4f46\u662f\uff0c\u76f8\u6bd4\u4e8e\u63d2\u5165\u548c\u63d2\u5165\u66f4\u65b0\u80fd\u4fdd\u8bc1\u6587\u4ef6\u5927\u5c0f\uff0c\u6279\u63d2\u5165\u5728\u8c03\u6574\u6587\u4ef6\u5927\u5c0f\u4e0a\u53ea\u80fd\u5c3d\u529b\u800c\u4e3a\u3002<\/li>\n<\/ul>\n

    DeltaStreamerPermalink<\/span><\/i><\/h2>\n

    HoodieDeltaStreamer\u5b9e\u7528\u5de5\u5177 (hudi-utilities-bundle\u4e2d\u7684\u4e00\u90e8\u5206) \u63d0\u4f9b\u4e86\u4eceDFS\u6216Kafka\u7b49\u4e0d\u540c\u6765\u6e90\u8fdb\u884c\u6444\u53d6\u7684\u65b9\u5f0f\uff0c\u5e76\u5177\u6709\u4ee5\u4e0b\u529f\u80fd\u3002<\/p>\n

      \n
    • \u4eceKafka\u5355\u6b21\u6444\u53d6\u65b0\u4e8b\u4ef6\uff0c\u4eceSqoop\u3001HiveIncrementalPuller\u8f93\u51fa\u6216DFS\u6587\u4ef6\u5939\u4e2d\u7684\u591a\u4e2a\u6587\u4ef6\u00a0\u589e\u91cf\u5bfc\u5165<\/li>\n
    • \u652f\u6301json\u3001avro\u6216\u81ea\u5b9a\u4e49\u8bb0\u5f55\u7c7b\u578b\u7684\u4f20\u5165\u6570\u636e<\/li>\n
    • \u7ba1\u7406\u68c0\u67e5\u70b9\uff0c\u56de\u6eda\u548c\u6062\u590d<\/li>\n
    • \u5229\u7528DFS\u6216Confluent\u00a0schema\u6ce8\u518c\u8868\u7684Avro\u6a21\u5f0f\u3002<\/li>\n
    • \u652f\u6301\u81ea\u5b9a\u4e49\u8f6c\u6362\u64cd\u4f5c<\/li>\n<\/ul>\n

      \u547d\u4ee4\u884c\u9009\u9879\u66f4\u8be6\u7ec6\u5730\u63cf\u8ff0\u4e86\u8fd9\u4e9b\u529f\u80fd\uff1a<\/p>\n

      \n
      \n
      [hoodie]$ spark-submit --class org.apache.hudi.utilities.deltastreamer.HoodieDeltaStreamer `ls packaging\/hudi-utilities-bundle\/target\/hudi-utilities-bundle-*.jar` --help\nUsage: <main class> [options]\n  Options:\n    --commit-on-errors\n        Commit even when some records failed to be written\n      Default: false\n    --enable-hive-sync\n          Enable syncing to hive\n       Default: false\n    --filter-dupes\n          Should duplicate records from source be dropped\/filtered outbefore\n          insert\/bulk-insert\n      Default: false\n    --help, -h\n    --hudi-conf\n          Any configuration that can be set in the properties file (using the CLI\n          parameter \\\"--propsFilePath\\\") can also be passed command line using this\n          parameter\n          Default: []\n    --op\n      Takes one of these values : UPSERT (default), INSERT (use when input is\n      purely new data\/inserts to gain speed)\n      Default: UPSERT\n      Possible Values: [UPSERT, INSERT, BULK_INSERT]\n    --payload-class\n      subclass of HoodieRecordPayload, that works off a GenericRecord.\n      Implement your own, if you want to do something other than overwriting\n      existing value\n      Default: org.apache.hudi.OverwriteWithLatestAvroPayload\n    --props\n      path to properties file on localfs or dfs, with configurations for\n      Hudi client, schema provider, key generator and data source. For\n      Hudi client props, sane defaults are used, but recommend use to\n      provide basic things like metrics endpoints, hive configs etc. For\n      sources, referto individual classes, for supported properties.\n      Default: file:\/\/\/Users\/vinoth\/bin\/hoodie\/src\/test\/resources\/delta-streamer-config\/dfs-source.properties\n    --schemaprovider-class\n      subclass of org.apache.hudi.utilities.schema.SchemaProvider to attach\n      schemas to input & target table data, built in options:\n      FilebasedSchemaProvider\n      Default: org.apache.hudi.utilities.schema.FilebasedSchemaProvider\n    --source-class\n      Subclass of org.apache.hudi.utilities.sources to read data. Built-in\n      options: org.apache.hudi.utilities.sources.{JsonDFSSource (default),\n      AvroDFSSource, JsonKafkaSource, AvroKafkaSource, HiveIncrPullSource}\n      Default: org.apache.hudi.utilities.sources.JsonDFSSource\n    --source-limit\n      Maximum amount of data to read from source. Default: No limit For e.g:\n      DFSSource => max bytes to read, KafkaSource => max events to read\n      Default: 9223372036854775807\n    --source-ordering-field\n      Field within source record to decide how to break ties between records\n      with same key in input data. Default: \\'ts\\' holding unix timestamp of\n      record\n      Default: ts\n    --spark-master\n      spark master to use.\n      Default: local[2]\n  * --target-base-path\n      base path for the target Hudi dataset. (Will be created if did not\n      exist first time around. If exists, expected to be a Hudi dataset)\n  * --target-table\n      name of the target table in Hive\n    --transformer-class\n      subclass of org.apache.hudi.utilities.transform.Transformer. UDF to\n      transform raw source dataset to a target dataset (conforming to target\n      schema) before writing. Default : Not set. E:g -\n      org.apache.hudi.utilities.transform.SqlQueryBasedTransformer (which\n      allows a SQL query template to be passed as a transformation function)\n<\/pre>\n<\/div>\n<\/div>\n
      \u8be5\u5de5\u5177\u91c7\u7528\u5c42\u6b21\u7ed3\u6784\u7ec4\u6210\u7684\u5c5e\u6027\u6587\u4ef6\uff0c\u5e76\u5177\u6709\u53ef\u63d2\u62d4\u7684\u63a5\u53e3\uff0c\u7528\u4e8e\u63d0\u53d6\u6570\u636e\u3001\u751f\u6210\u5bc6\u94a5\u548c\u63d0\u4f9b\u6a21\u5f0f\u3002 \u4eceKafka\u548cDFS\u6444\u53d6\u6570\u636e\u7684\u793a\u4f8b\u914d\u7f6e\u5728\u8fd9\u91cc\uff1ahudi-utilities\/src\/test\/resources\/delta-streamer-config\u3002
      \n\u4f8b\u5982\uff1a\u5f53\u60a8\u8ba9Confluent Kafka\u3001Schema\u6ce8\u518c\u8868\u542f\u52a8\u5e76\u8fd0\u884c\u540e\uff0c\u53ef\u4ee5\u7528\u8fd9\u4e2a\u547d\u4ee4\u4ea7\u751f\u4e00\u4e9b\u6d4b\u8bd5\u6570\u636e \uff08impressions.avro\uff0c \u7531schema-registry\u4ee3\u7801\u5e93\u63d0\u4f9b\uff09<\/p>\n
      \n
      \n
      [confluent-5.0.0]$ bin\/ksql-datagen schema=..\/impressions.avro format=avro topic=impressions key=impressionid\n<\/pre>\n<\/div>\n<\/div>\n
      \u7136\u540e\u7528\u5982\u4e0b\u547d\u4ee4\u6444\u53d6\u8fd9\u4e9b\u6570\u636e\u3002<\/p>\n
      \n
      \n
      [hoodie]$ spark-submit --class org.apache.hudi.utilities.deltastreamer.HoodieDeltaStreamer `ls packaging\/hudi-utilities-bundle\/target\/hudi-utilities-bundle-*.jar` \\\\\n  --props file:\/\/${PWD}\/hudi-utilities\/src\/test\/resources\/delta-streamer-config\/kafka-source.properties \\\\\n  --schemaprovider-class org.apache.hudi.utilities.schema.SchemaRegistryProvider \\\\\n  --source-class org.apache.hudi.utilities.sources.AvroKafkaSource \\\\\n  --source-ordering-field impresssiontime \\\\\n  --target-base-path file:\/\/\/tmp\/hudi-deltastreamer-op --target-table uber.impressions \\\\\n  --op BULK_INSERT\n<\/pre>\n<\/div>\n<\/div>\n
       <\/p>\n
      Datasource WriterPermalink<\/span><\/i><\/h2>\n
      hudi-spark\u6a21\u5757\u63d0\u4f9b\u4e86DataSource API\uff0c\u53ef\u4ee5\u5c06\u4efb\u4f55DataFrame\u5199\u5165\uff08\u4e5f\u53ef\u4ee5\u8bfb\u53d6\uff09\u5230Hudi\u6570\u636e\u96c6\u4e2d\u3002 \u4ee5\u4e0b\u662f\u5728\u6307\u5b9a\u9700\u8981\u4f7f\u7528\u7684\u5b57\u6bb5\u540d\u79f0\u7684\u4e4b\u540e\uff0c\u5982\u4f55\u63d2\u5165\u66f4\u65b0DataFrame\u7684\u65b9\u6cd5\uff0c\u8fd9\u4e9b\u5b57\u6bb5\u5305\u62ec\u00a0recordKey => _row_key\u3001partitionPath => partition\u548cprecombineKey => timestamp<\/p>\n
      \n
      \n
      inputDF.write()\n       .format(\\\"org.apache.hudi\\\")\n       .options(clientOpts) \/\/ \u53ef\u4ee5\u4f20\u5165\u4efb\u4f55Hudi\u5ba2\u6237\u7aef\u53c2\u6570\n       .option(DataSourceWriteOptions.RECORDKEY_FIELD_OPT_KEY(), \\\"_row_key\\\")\n       .option(DataSourceWriteOptions.PARTITIONPATH_FIELD_OPT_KEY(), \\\"partition\\\")\n       .option(DataSourceWriteOptions.PRECOMBINE_FIELD_OPT_KEY(), \\\"timestamp\\\")\n       .option(HoodieWriteConfig.TABLE_NAME, tableName)\n       .mode(SaveMode.Append)\n       .save(basePath);\n<\/pre>\n<\/div>\n<\/div>\n
      \u4e0eHive\u540c\u6b65Permalink<\/i><\/h2>\n
      \u4e0a\u9762\u7684\u4e24\u4e2a\u5de5\u5177\u90fd\u652f\u6301\u5c06\u6570\u636e\u96c6\u7684\u6700\u65b0\u6a21\u5f0f\u540c\u6b65\u5230Hive Metastore\uff0c\u4ee5\u4fbf\u67e5\u8be2\u65b0\u7684\u5217\u548c\u5206\u533a\u3002 \u5982\u679c\u9700\u8981\u4ece\u547d\u4ee4\u884c\u6216\u5728\u72ec\u7acb\u7684JVM\u4e2d\u8fd0\u884c\u5b83\uff0cHudi\u63d0\u4f9b\u4e86\u4e00\u4e2aHiveSyncTool\uff0c \u5728\u6784\u5efa\u4e86hudi-hive\u6a21\u5757\u4e4b\u540e\uff0c\u53ef\u4ee5\u6309\u4ee5\u4e0b\u65b9\u5f0f\u8c03\u7528\u5b83\u3002<\/p>\n
      \n
      \n
      cd hudi-hive\n.\/run_sync_tool.sh\n [hudi-hive]$ .\/run_sync_tool.sh --help\nUsage: <main class> [options]\n  Options:\n  * --base-path\n       Basepath of Hudi dataset to sync\n  * --database\n       name of the target database in Hive\n    --help, -h\n       Default: false\n  * --jdbc-url\n       Hive jdbc connect url\n  * --pass\n       Hive password\n  * --table\n       name of the target table in Hive\n  * --user\n       Hive username<\/pre>\n<\/div>\n<\/div>\n
      \u5220\u9664\u6570\u636ePermalink<\/span><\/i><\/h2>\n
      \u901a\u8fc7\u5141\u8bb8\u7528\u6237\u6307\u5b9a\u4e0d\u540c\u7684\u6570\u636e\u8bb0\u5f55\u8d1f\u8f7d\u5b9e\u73b0\uff0cHudi\u652f\u6301\u5bf9\u5b58\u50a8\u5728Hudi\u6570\u636e\u96c6\u4e2d\u7684\u6570\u636e\u6267\u884c\u4e24\u79cd\u7c7b\u578b\u7684\u5220\u9664\u3002<\/p>\n