流配置示例

这里列出一些流配置示例以做参考。

读Pulsar写Kafka

一般流配置示例如下:

  1. {
  2. "id": 1,
  3. "source_info": {
  4. "type": "pulsar",
  5. "admin_url": "YOUR_PULSAR_ADMIN_URL",
  6. "service_url": "YOUR_PULSAR_SERVICE_URL",
  7. "topic": "YOUR_PULSAR_TOPIC",
  8. "subscription_name": "debezium2canal",
  9. "deserialization_info": {
  10. "type": "debezium_json",
  11. "ignore_parse_errors": true,
  12. "timestamp_format_standard": "ISO_8601"
  13. },
  14. "fields": [
  15. {
  16. "type": "base",
  17. "name": "name",
  18. "format_info": {
  19. "type": "string"
  20. }
  21. },
  22. {
  23. "type": "base",
  24. "name": "age",
  25. "format_info": {
  26. "type": "int"
  27. }
  28. }
  29. ],
  30. "authentication": null
  31. },
  32. "sink_info": {
  33. "type": "kafka",
  34. "fields": [
  35. {
  36. "type": "base",
  37. "name": "name",
  38. "format_info": {
  39. "type": "string"
  40. }
  41. },
  42. {
  43. "type": "base",
  44. "name": "age",
  45. "format_info": {
  46. "type": "int"
  47. }
  48. }
  49. ],
  50. "address": "YOUR_KAFKA_ADDRESS",
  51. "topic": "sort_test_canal",
  52. "serialization_info": {
  53. "type": "canal"
  54. }
  55. },
  56. "properties": {
  57. "consumer.bootstrap-mode": "earliest",
  58. "transaction.timeout.ms": 900000
  59. }
  60. }

整库迁移流配置示例:

  1. {
  2. "id": 123,
  3. "source_info": {
  4. "type": "pulsar",
  5. "admin_url": "YOUR_PULSAR_ADMIN_URL",
  6. "service_url": "YOUR_PULSAR_SERVICE_URL",
  7. "topic": "YOUR_PULSAR_TOPIC",
  8. "subscription_name": "whole-db-migration",
  9. "deserialization_info": {
  10. "type": "debezium_json",
  11. "ignore_parse_errors": false,
  12. "timestamp_format_standard": "ISO_8601",
  13. "include_update_before": true
  14. },
  15. "fields": [
  16. {
  17. "type": "builtin",
  18. "name": "db",
  19. "format_info": {
  20. "type": "string"
  21. },
  22. "builtin_field": "MYSQL_METADATA_DATABASE"
  23. },
  24. {
  25. "type": "builtin",
  26. "name": "table",
  27. "format_info": {
  28. "type": "string"
  29. },
  30. "builtin_field": "MYSQL_METADATA_TABLE"
  31. },
  32. {
  33. "type": "builtin",
  34. "name": "mydata",
  35. "format_info": {
  36. "type": "string"
  37. },
  38. "builtin_field": "MYSQL_METADATA_DATA"
  39. },
  40. {
  41. "type": "builtin",
  42. "name": "es",
  43. "format_info": {
  44. "type": "long"
  45. },
  46. "builtin_field": "MYSQL_METADATA_EVENT_TIME"
  47. },
  48. {
  49. "type": "builtin",
  50. "name": "isDdl",
  51. "format_info": {
  52. "type": "boolean"
  53. },
  54. "builtin_field": "MYSQL_METADATA_IS_DDL"
  55. },
  56. {
  57. "type": "builtin",
  58. "name": "type",
  59. "format_info": {
  60. "type": "string"
  61. },
  62. "builtin_field": "MYSQL_METADATA_EVENT_TYPE"
  63. }
  64. ],
  65. "authentication": null
  66. },
  67. "sink_info": {
  68. "type": "kafka",
  69. "fields": [
  70. {
  71. "type": "builtin",
  72. "name": "db",
  73. "format_info": {
  74. "type": "string"
  75. },
  76. "builtin_field": "MYSQL_METADATA_DATABASE"
  77. },
  78. {
  79. "type": "builtin",
  80. "name": "table",
  81. "format_info": {
  82. "type": "string"
  83. },
  84. "builtin_field": "MYSQL_METADATA_TABLE"
  85. },
  86. {
  87. "type": "builtin",
  88. "name": "mydata",
  89. "format_info": {
  90. "type": "string"
  91. },
  92. "builtin_field": "MYSQL_METADATA_DATA"
  93. },
  94. {
  95. "type": "builtin",
  96. "name": "es",
  97. "format_info": {
  98. "type": "long"
  99. },
  100. "builtin_field": "MYSQL_METADATA_EVENT_TIME"
  101. },
  102. {
  103. "type": "builtin",
  104. "name": "isDdl",
  105. "format_info": {
  106. "type": "boolean"
  107. },
  108. "builtin_field": "MYSQL_METADATA_IS_DDL"
  109. },
  110. {
  111. "type": "builtin",
  112. "name": "type",
  113. "format_info": {
  114. "type": "string"
  115. },
  116. "builtin_field": "MYSQL_METADATA_EVENT_TYPE"
  117. }
  118. ],
  119. "address": "YOUR_KAFKA_ADDRESS",
  120. "topic": "whole-db-migration",
  121. "serialization_info": {
  122. "type": "canal"
  123. }
  124. },
  125. "properties": {
  126. "transaction.timeout.ms": 900000,
  127. "consumer.bootstrap-mode": "earliest"
  128. }
  129. }

读Pulsar写Hive

一般流配置示例:

  1. {
  2. "id": 123,
  3. "source_info": {
  4. "type": "pulsar",
  5. "admin_url": "http://100.76.43.216:8080",
  6. "service_url": "pulsar://100.76.43.216:6650",
  7. "topic": "persistent://public/public/b_pzr",
  8. "subscription_name": "whole-db-migration",
  9. "deserialization_info": {
  10. "type": "debezium_json",
  11. "ignore_parse_errors": false,
  12. "timestamp_format_standard": "ISO_8601",
  13. "include_update_before": true
  14. },
  15. "fields": [
  16. {
  17. "type": "base",
  18. "name": "f1",
  19. "format_info": {
  20. "type": "string"
  21. }
  22. },
  23. {
  24. "type": "base",
  25. "name": "f2",
  26. "format_info": {
  27. "type": "int"
  28. }
  29. },
  30. {
  31. "type": "builtin",
  32. "name": "data_time",
  33. "format_info": {
  34. "type": "string"
  35. },
  36. "builtin_field": "DATA_TIME"
  37. }
  38. ],
  39. "authentication": null
  40. },
  41. "sink_info": {
  42. "type": "hive",
  43. "fields": [
  44. {
  45. "type": "base",
  46. "name": "f1",
  47. "format_info": {
  48. "type": "string"
  49. }
  50. },
  51. {
  52. "type": "base",
  53. "name": "f2",
  54. "format_info": {
  55. "type": "int"
  56. }
  57. },
  58. {
  59. "type": "builtin",
  60. "name": "data_time",
  61. "format_info": {
  62. "type": "string"
  63. },
  64. "builtin_field": "DATA_TIME"
  65. }
  66. ],
  67. "hive_server_jdbc_url": "YOUR_HIVE_JDBC_URL",
  68. "database": "YOUR_HIVE_DB_NAME",
  69. "table": "YOUR_HIVE_TABLE_NAME",
  70. "username": "username",
  71. "password": "password",
  72. "data_path": "YOUR_HIVE_TABLE_DATA_PATH_ON_HDFS",
  73. "partitions": [
  74. {
  75. "type": "time",
  76. "field_name": "data_time",
  77. "date_format": "yyyy-MM-dd"
  78. },
  79. {
  80. "type": "field",
  81. "field_name": "f2"
  82. }
  83. ],
  84. "file_format": {
  85. "type": "text",
  86. "splitter": "|",
  87. "compression_type": "GZIP"
  88. },
  89. "hadoop_proxy_user": "proxyUser"
  90. },
  91. "properties": {
  92. "transaction.timeout.ms": 900000,
  93. "consumer.bootstrap-mode": "latest"
  94. }
  95. }