DataX迁移MongoDB_datax 迁移mongodb

DataX迁移MongoDB

• 项目地址：GitHub - alibaba/DataX: DataX是阿里云DataWorks数据集成的开源版本。
• 迁移MongoDB，读取组件为mongodbreader，写入组件为mongodbwriter

源码修改

• 目前版本中，在迁移MongoDB时，若列的类型为二进制，mongodbreader未做处理，源码src/main/java/com/alibaba/datax/plugin/reader/mongodbreader/MongoDBReader.java

  if (tempCol == null) {
      //continue; 这个不能直接continue会导致record到目的端错位
      record.addColumn(new StringColumn(null));
  }else if (tempCol instanceof Double) {
      //TODO deal with Double.isNaN()
      record.addColumn(new DoubleColumn((Double) tempCol));
  } else if (tempCol instanceof Boolean) {
      record.addColumn(new BoolColumn((Boolean) tempCol));
  } else if (tempCol instanceof Date) {
      record.addColumn(new DateColumn((Date) tempCol));
  } else if (tempCol instanceof Integer) {
      record.addColumn(new LongColumn((Integer) tempCol));
  }else if (tempCol instanceof Long) {
      record.addColumn(new LongColumn((Long) tempCol));
  } else {
      if(KeyConstant.isArrayType(column.getString(KeyConstant.COLUMN_TYPE))) {
          String splitter = column.getString(KeyConstant.COLUMN_SPLITTER);
          if(Strings.isNullOrEmpty(splitter)) {
              throw DataXException.asDataXException(MongoDBReaderErrorCode.ILLEGAL_VALUE,
                  MongoDBReaderErrorCode.ILLEGAL_VALUE.getDescription());
          } else {
              ArrayList array = (ArrayList)tempCol;
              String tempArrayStr = Joiner.on(splitter).join(array);
              record.addColumn(new StringColumn(tempArrayStr));
          }
      } else {
          record.addColumn(new StringColumn(tempCol.toString()));
      }
  }
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12
  13
  14
  15
  16
  17
  18
  19
  20
  21
  22
  23
  24
  25
  26
  27
  28
  29

• 修改为：

  if (tempCol == null) {
      //continue; 这个不能直接continue会导致record到目的端错位
      record.addColumn(new StringColumn(null));
  }else if (tempCol instanceof Double) {
      //TODO deal with Double.isNaN()
      record.addColumn(new DoubleColumn((Double) tempCol));
  } else if (tempCol instanceof Boolean) {
      record.addColumn(new BoolColumn((Boolean) tempCol));
  } else if (tempCol instanceof Date) {
      record.addColumn(new DateColumn((Date) tempCol));
  } else if (tempCol instanceof Integer) {
      record.addColumn(new LongColumn((Integer) tempCol));
  }else if (tempCol instanceof Long) {
      record.addColumn(new LongColumn((Long) tempCol));
  }  else if (tempCol instanceof Binary) {
      // 处理 MongoDB 的 Binary 类型数据
      Binary binaryData = (Binary) tempCol;
      byte[] binaryBytes = binaryData.getData();
      // 将字节数组添加到 DataX 中的二进制列
      record.addColumn(new BytesColumn(binaryBytes));
  } else {
      if(KeyConstant.isArrayType(column.getString(KeyConstant.COLUMN_TYPE))) {
          String splitter = column.getString(KeyConstant.COLUMN_SPLITTER);
          if(Strings.isNullOrEmpty(splitter)) {
              throw DataXException.asDataXException(MongoDBReaderErrorCode.ILLEGAL_VALUE,
                  MongoDBReaderErrorCode.ILLEGAL_VALUE.getDescription());
          } else {
              ArrayList array = (ArrayList)tempCol;
              String tempArrayStr = Joiner.on(splitter).join(array);
              record.addColumn(new StringColumn(tempArrayStr));
          }
      } else {
          record.addColumn(new StringColumn(tempCol.toString()));
      }
  }
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12
  13
  14
  15
  16
  17
  18
  19
  20
  21
  22
  23
  24
  25
  26
  27
  28
  29
  30
  31
  32
  33
  34
  35

  • 修改源码后，要重新打包，由于只更改了mongodbreader，故在打包时，可以考虑将根

迁移脚本

• 编写job脚本：1.json

  {
      "job": {
          "content": [
              {
                  "reader": {
                      "name": "mongodbreader",
                      "parameter": {
                          "address": ["ip1:27017"],
                          "collectionName": "data",
                          "column": [
  							{
                                  "name": "_id",
                                  "type": "long"
                              },
                              {
                                  "name": "fileContent",
                                  "type": "bytes"
                              }
  						],
                          "dbName": "monitor",
                          "userName": "root",
                          "userPassword": "123456",
  						"query": {
  						  "_id": {
  							"$lt": 21
  						  }
  						}
                      }
                  },
                  "writer": {
                      "name": "mongodbwriter",
                      "parameter": {
                          "address": ["ip2:27017"],
                          "collectionName": "data",
                          "column": [
  							{
                                  "name": "_id",
                                  "type": "long"
                              },
                              {
                                  "name": "fileContent",
                                  "type": "bytes"
                              }
  						],
  						"writeMode": {
  						  "isReplace": "true",
  						  "replaceKey": "_id"
  						}
                          "dbName": "test",
  						"userName": "root",
                          "userPassword": "123456",
                      }
                  }
              }
          ],
          "setting": {
              "speed": {
                  "channel": "2"
              }
          }
      }
  }
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12
  13
  14
  15
  16
  17
  18
  19
  20
  21
  22
  23
  24
  25
  26
  27
  28
  29
  30
  31
  32
  33
  34
  35
  36
  37
  38
  39
  40
  41
  42
  43
  44
  45
  46
  47
  48
  49
  50
  51
  52
  53
  54
  55
  56
  57
  58
  59
  60
  61
  62

  • reader中的query节点为查询条件，上述demo中是查询_id小于21的记录。

• 执行命令：

  python datax.py G:\Code\1.json
  1

  • datax.py在打包后的target目录下，相对路径：target\datax\datax\bin