CDCUtils.java 2.7 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687
  1. package com.saga.flink.cdc.utils;
  2. import com.alibaba.fastjson.JSONObject;
  3. import com.saga.flink.cdc.constant.CDCConstant;
  4. import com.saga.flink.cdc.constant.OpEnum;
  5. import org.apache.kafka.connect.data.Field;
  6. import org.apache.kafka.connect.data.Struct;
  7. import org.apache.kafka.connect.source.SourceRecord;
  8. import java.util.List;
  9. public class CDCUtils {
  10. public static JSONObject getCDCDataJSON(SourceRecord sourceRecord){
  11. Struct dataRecord = (Struct) sourceRecord.value();
  12. JSONObject cdcJson = new JSONObject();
  13. // 判断数据操作
  14. String op = dataRecord.getString(CDCConstant.OP);
  15. String op_type = OpEnum.getValue(op);
  16. //未知操作
  17. if ("".equals(op_type)){
  18. op_type = op;
  19. }
  20. cdcJson.put(CDCConstant.OP, op_type);
  21. Struct source = dataRecord.getStruct(CDCConstant.SOURCE);
  22. Object db = source.get(CDCConstant.DB);
  23. Object table = source.get(CDCConstant.TABLE);
  24. Object ts_ms = source.get(CDCConstant.TS_MS);
  25. cdcJson.put(CDCConstant.DB, db);
  26. cdcJson.put(CDCConstant.TABLE, table);
  27. cdcJson.put(CDCConstant.TS_MS, ts_ms);
  28. if (!CDCConstant.OP_D.equals(op)){
  29. Struct after = dataRecord.getStruct(CDCConstant.AFTER);
  30. JSONObject afterJson = fieldToJson(after);
  31. cdcJson.put(CDCConstant.AFTER, afterJson);
  32. }
  33. if (CDCConstant.OP_U.equals(op) || CDCConstant.OP_D.equals(op)){
  34. Struct before = dataRecord.getStruct(CDCConstant.BEFORE);
  35. JSONObject beforeJson = fieldToJson(before);
  36. cdcJson.put(CDCConstant.BEFORE, beforeJson);
  37. }
  38. // 主键 json ,kafka 分区 key
  39. JSONObject pkJson = new JSONObject();
  40. Struct pkey = (Struct) sourceRecord.key();
  41. List<Field> fields = pkey.schema().fields();
  42. Integer hash = 0;
  43. for (Field field : fields) {
  44. String n = field.name();
  45. Object v = pkey.get(n);
  46. pkJson.put(n, v);
  47. hash += v.hashCode();
  48. }
  49. Integer partitionNum = Math.abs(hash) % CDCConstant.KAFKA_PARTITION_NUM;
  50. cdcJson.put(CDCConstant.KAFKA_PARTITION_KEY, partitionNum);
  51. cdcJson.put(CDCConstant.PRIMARY_JSON, pkJson);
  52. return cdcJson;
  53. }
  54. private static JSONObject fieldToJson(Struct struct){
  55. if (struct == null){
  56. return new JSONObject();
  57. }
  58. List<Field> fields = struct.schema().fields();
  59. JSONObject json = new JSONObject();
  60. for (Field field : fields) {
  61. String fieldName = field.name();
  62. Object fieldValue = struct.get(fieldName);
  63. json.put(fieldName, fieldValue);
  64. }
  65. return json;
  66. }
  67. }