通用类

CDM有哪些优势？

CDM有哪些安全防护？

CDM是一个完全托管的服务，提供了以下安全防护能力保护用户数据安全。

• 实例隔离：CDM服务的用户只能使用自己创建的实例，实例和实例之间是相互隔离的，不可相互访问。

• 系统加固：CDM实例的操作系统进行了特别的安全加固，攻击者无法从Internet访问CDM实例的操作系统。

• 密钥加密：用户在CDM上创建连接输入的各种数据源的密钥，CDM均采用高强度加密算法保存在CDM数据库。

• 无中间存储：数据在迁移的过程中，CDM只处理数据映射和转换，而不会存储任何用户数据或片段。

如何降低CDM使用成本？

如果是迁移公网的数据上云，可以使用NAT网关服务，实现CDM服务与子网中的其他弹性云主机共享弹性IP，可以更经济、更方便的通过Internet迁移本地数据中心或第三方云上的数据。

具体操作如下：

1. 假设已经创建好了CDM集群（无需为CDM集群绑定专用弹性IP），记录下CDM集群所在的VPC和子网。

2. 创建NAT网关，注意选择和CDM集群相同的VPC、子网。

3. 创建完NAT网关后，回到NAT网关控制台列表，单击创建好的网关名称，然后选择“添加SNAT规则”。

4. 选择子网和弹性IP，如果没有弹性IP，需要先申请一个。

   完成之后，就可以到CDM控制台，通过Internet迁移公网的数据上云了。例如：迁移本地数据中心FTP服务器上的文件到OBS、迁移第三方云上关系型数据库到云服务RDS。

说明如果用户对本地数据源的访问通道做了SSL加密，则CDM无法通过弹性IP连接数据源。

CDM集群是否支持升级操作？

不支持。

CDM集群目前暂不支持升级操作，如果需要使用高版本集群则需要重新创建。

CDM迁移性能如何？

单个cdm.large规格实例理论上可以支持1TB～8TB/天的数据迁移。

但在实际使用中，实际传输速率受公网带宽、集群规格、文件读写速度、作业并发数设置、磁盘读写性能等因素影响。

CDM不同集群规格对应并发的作业数是多少？

CDM不同集群规格对应并发的作业数如表下表所示。

包含但不限于以下情况，建议使用多个CDM集群进行业务分流：

• 作为不同的用途，例如用于数据迁移作业，或作为DataArts Studio管理中心连接代理。

• 给不同的业务部门使用，例如财务、网上商城等。

功能类

是否支持增量迁移？

CDM支持增量数据迁移。利用定时任务配置和时间宏变量函数等参数，可支持以下场景的增量数据迁移：

• 文件增量迁移

• 关系数据库增量迁移

• 使用时间宏变量完成增量同步

• HBase/CloudTable增量迁移

是否支持字段转换？

支持，CDM支持以下字段转换器：

• 脱敏

• 去前后空格

• 字符串反转

• 字符串替换

• 表达式转换

  在创建表/文件迁移作业的字段映射界面，可新建字段转换器，如图下图所示。

脱敏

隐藏字符串中的关键信息，例如要将“12345678910”转换为“123***8910”，则配置如下：

• “起始保留长度”为“3”。

• “结尾保留长度”为“4”。

• “替换字符”为“”。

如下图“字段脱敏”：

去前后空格

自动去字符串前后的空值，不需要配置参数。

字符串反转

自动反转字符串，例如将“ABC”转换为“CBA”，不需要配置参数。

字符串替换

替换字符串，需要用户配置被替换的对象，以及替换后的值。

表达式转换

使用JSP表达式语言（Expression Language）对当前字段或整行数据进行转换。JSP表达式语言可以用来创建算术和逻辑表达式。在表达式内可以使用整型数，浮点数，字符串，常量true、false和null。

表达式支持以下两个环境变量：

• value：当前字段值。

• row：当前行，数组类型。

表达式支持以下工具类：

• StringUtils：字符串处理类，参考Java SDK代码的包结构“org.apache.commons.lang.StringUtils”。

• DateUtils：日期工具类。

• CommonUtils：公共工具类。

• NumberUtils：字符串转数值类。

• HttpsUtils：读取网络文件类。

应用举例：

1. 如果当前字段为字符串类型，将字符串全部转换为小写，例如将“aBC”转换为“abc”。
   表达式：StringUtils.lowerCase(value)

2. 将当前字段的字符串全部转为大写。
   表达式：StringUtils.upperCase(value)

3. 如果当前字段值为“yyyy-MM-dd”格式的日期字符串，需要截取年，例如字段值为“2017-12-01”，转换后为“2017”。
   表达式：StringUtils.substringBefore(value,"-")

4. 如果当前字段值为数值类型，转换后值为当前值的两倍。
   表达式：value*2

5. 如果当前字段值为“true”，转换后为“Y”，其它值则转换后为“N”。
   表达式：value=="true"?"Y":"N"

6. 如果当前字段值为字符串类型，当为空时，转换为“Default”，否则不转换。
   表达式：empty value? "Default":value

7. 如果想将日期字段格式从“2018/01/05 15:15:05”转换为“2018-01-05 15:15:05”。
   表达式：DateUtils.format(DateUtils.parseDate(value,"yyyy/MM/dd HH:mm:ss"),"yyyy-MM-dd HH:mm:ss")

8. 获取一个36位的UUID（Universally Unique Identifier，通用唯一识别码）。
   表达式：CommonUtils.randomUUID()

9. 如果当前字段值为字符串类型，将首字母转换为大写，例如将“cat”转换为“Cat”。
   表达式：StringUtils.capitalize(value)

10. 如果当前字段值为字符串类型，将首字母转换为小写，例如将“Cat”转换为“cat”。
    表达式：StringUtils.uncapitalize(value)

11. 如果当前字段值为字符串类型，使用空格填充为指定长度，并且将字符串居中，当字符串长度不小于指定长度时不转换，例如将“ab”转换为长度为4的“ab”。
    表达式：StringUtils.center(value,4)

12. 删除字符串末尾的一个换行符（包括“\n”、“\r”或者“\r\n”），例如将“abc\r\n\r\n”转换为“abc\r\n”。
    表达式：StringUtils.chomp(value)

13. 如果字符串中包含指定的字符串，则返回布尔值true，否则返回false。例如“abc”中包含“a”，则返回true。
    表达式：StringUtils.contains(value,"a")

14. 如果字符串中包含指定字符串的任一字符，则返回布尔值true，否则返回false。例如“zzabyycdxx”中包含“z”或“a”任意一个，则返回true。
    表达式：StringUtils.containsAny("value","za")

15. 如果字符串中不包含指定的所有字符，则返回布尔值true，包含任意一个字符则返回false。例如“abz”中包含“xyz”里的任意一个字符，则返回false。
    表达式：StringUtils.containsNone(value,"xyz")

16. 如果当前字符串只包含指定字符串中的字符，则返回布尔值true，包含任意一个其它字符则返回false。例如“abab”只包含“abc”中的字符，则返回true。
    表达式：StringUtils.containsOnly(value,"abc")

17. 如果字符串为空或null，则转换为指定的字符串，否则不转换。例如将空字符转换为null。
    表达式：StringUtils.defaultIfEmpty(value,null)

18. 如果字符串以指定的后缀结尾（包括大小写），则返回布尔值true，否则返回false。例如“abcdef”后缀不为null，则返回false。
    表达式：StringUtils.endsWith(value,null)

19. 如果字符串和指定的字符串完全一样（包括大小写），则返回布尔值true，否则返回false。例如比较字符串“abc”和“ABC”，则返回false。
    表达式：StringUtils.equals(value,"ABC")

20. 从字符串中获取指定字符串的第一个索引，没有则返回整数-1。例如从“aabaabaa”中获取“ab”的第一个索引1。
    表达式：StringUtils.indexOf(value,"ab")

21. 从字符串中获取指定字符串的最后一个索引，没有则返回整数-1。例如从“aFkyk”中获取“k”的最后一个索引4。
    表达式：StringUtils.lastIndexOf(value,"k")

22. 从字符串中指定的位置往后查找，获取指定字符串的第一个索引，没有则转换为“-1”。例如“aabaabaa”中索引3的后面，第一个“b”的索引是5。
    表达式：StringUtils.indexOf(value,"b",3)

23. 从字符串获取指定字符串中任一字符的第一个索引，没有则返回整数-1。例如从“zzabyycdxx”中获取“z”或“a”的第一个索引0。
    表达式：StringUtils.indexOfAny(value,"za")

24. 如果字符串仅包含Unicode字符，返回布尔值true，否则返回false。例如“ab2c”中包含非Unicode字符，返回false。
    表达式：StringUtils.isAlpha(value)

25. 如果字符串仅包含Unicode字符或数字，返回布尔值true，否则返回false。例如“ab2c”中仅包含Unicode字符和数字，返回true。
    表达式：StringUtils.isAlphanumeric(value)

26. 如果字符串仅包含Unicode字符、数字或空格，返回布尔值true，否则返回false。例如“ab2c”中仅包含Unicode字符和数字，返回true。
    表达式：StringUtils.isAlphanumericSpace(value)

27. 如果字符串仅包含Unicode字符或空格，返回布尔值true，否则返回false。例如“ab2c”中包含Unicode字符和数字，返回false。
    表达式：StringUtils.isAlphaSpace(value)

28. 如果字符串仅包含ASCII可打印字符，返回布尔值true，否则返回false。例如“!ab-c~”返回true。
    表达式：StringUtils.isAsciiPrintable(value)

29. 如果字符串为空或null，返回布尔值true，否则返回false。
    表达式：StringUtils.isEmpty(value)

30. 如果字符串中仅包含Unicode数字，返回布尔值true，否则返回false。
    表达式：StringUtils.isNumeric(value)

31. 获取字符串最左端的指定长度的字符，例如获取“abc”最左端的2位字符“ab”。
    表达式：StringUtils.left(value,2)

32. 获取字符串最右端的指定长度的字符，例如获取“abc”最右端的2位字符“bc”。
    表达式：StringUtils.right(value,2)

33. 将指定字符串拼接至当前字符串的左侧，需同时指定拼接后的字符串长度，如果当前字符串长度不小于指定长度，则不转换。例如将“yz”拼接到“bat”左侧，拼接后长度为8，则转换后为“yzyzybat”。
    表达式：StringUtils.leftPad(value,8,"yz")

34. 将指定字符串拼接至当前字符串的右侧，需同时指定拼接后的字符串长度，如果当前字符串长度不小于指定长度，则不转换。例如将“yz”拼接到“bat”右侧，拼接后长度为8，则转换后为“batyzyzy”。
    表达式：StringUtils.rightPad(value,8,"yz")

35. 如果当前字段为字符串类型，获取当前字符串的长度，如果该字符串为null，则返回0。
    表达式：StringUtils.length(value)

36. 如果当前字段为字符串类型，删除其中所有的指定字符串，例如从“queued”中删除“ue”，转换后为“qd”。
    表达式：StringUtils.remove(value,"ue")

37. 如果当前字段为字符串类型，移除当前字段末尾指定的子字符串。指定的子字符串若不在当前字段的末尾，则不转换，例如移除当前字段“www.ctyun.cn”后的“.cn”。
    表达式：StringUtils.removeEnd(value,".com")

38. 如果当前字段为字符串类型，移除当前字段开头指定的子字符串。指定的子字符串若不在当前字段的开头，则不转换，例如移除当前字段“www.ctyun.cn”前的“www.”。
    表达式：StringUtils.removeStart(value,"www.")

39. 如果当前字段为字符串类型，替换当前字段中所有的指定字符串，例如将“aba”中的“a”用“z”替换，转换后为“zbz”。
    表达式：StringUtils.replace(value,"a","z")

40. 如果当前字段为字符串类型，一次替换字符串中的多个字符，例如将字符串“hello”中的“h”用“j”替换，“o”用“y”替换，转换后为“jelly”。
    表达式：StringUtils.replaceChars(value,"ho","jy")

41. 如果字符串以指定的前缀开头（区分大小写），则返回布尔值true，否则返回false，例如当前字符串“abcdef”以“abc”开头，则返回true。
    表达式：StringUtils.startsWith(value,"abc")

42. 如果当前字段为字符串类型，去除字段中所有指定的字符，例如去除“abcyx”中所有的“x”、“y”和“z”，转换后为“abc”。
    表达式：StringUtils.strip(value,"xyz")

43. 如果当前字段为字符串类型，去除字段末尾所有指定的字符，例如去除当前字段末尾的所有空格。
    表达式：StringUtils.stripEnd(value,null)

44. 如果当前字段为字符串类型，去除字段开头所有指定的字符，例如去除当前字段开头的所有空格。
    表达式：StringUtils.stripStart(value,null)

45. 如果当前字段为字符串类型，获取字符串指定位置后（不包括指定位置的字符）的子字符串，指定位置如果为负数，则从末尾往前计算位置。例如获取“abcde”第2个字符后的字符串，则转换后为“cde”。
    表达式：StringUtils.substring(value,2)

46. 如果当前字段为字符串类型，获取字符串指定区间的子字符串，区间位置如果为负数，则从末尾往前计算位置。例如获取“abcde”第2个字符后、第5个字符前的字符串，则转换后为“cd”。
    表达式：StringUtils.substring(value,2,5)

47. 如果当前字段为字符串类型，获取当前字段里第一个指定字符后的子字符串。例如获取“abcba”中第一个“b”之后的子字符串，转换后为“cba”。
    表达式：StringUtils.substringAfter(value,"b")

48. 如果当前字段为字符串类型，获取当前字段里最后一个指定字符后的子字符串。例如获取“abcba”中最后一个“b”之后的子字符串，转换后为“a”。
    表达式：StringUtils.substringAfterLast(value,"b")

49. 如果当前字段为字符串类型，获取当前字段里第一个指定字符前的子字符串。例如获取“abcba”中第一个“b”之前的子字符串，转换后为“a”。
    表达式：StringUtils.substringBefore(value,"b")

50. 如果当前字段为字符串类型，获取当前字段里最后一个指定字符前的子字符串。例如获取“abcba”中最后一个“b”之前的子字符串，转换后为“abc”。
    表达式：StringUtils.substringBeforeLast(value,"b")

51. 如果当前字段为字符串类型，获取嵌套在指定字符串之间的子字符串，没有匹配的则返回null。例如获取“tagabctag”中“tag”之间的子字符串，转换后为“abc”。
    表达式：StringUtils.substringBetween(value,"tag")

52. 如果当前字段为字符串类型，删除当前字符串两端的控制字符（char≤32），例如删除字符串前后的空格。
    表达式：StringUtils.trim(value)

53. 将当前字符串转换为字节，如果转换失败，则返回0。
    表达式：NumberUtils.toByte(value)

54. 将当前字符串转换为字节，如果转换失败，则返回指定值，例如指定值配置为1。
    表达式：NumberUtils.toByte(value,1)

55. 将当前字符串转换为Double数值，如果转换失败，则返回0.0d。
    表达式：NumberUtils.toDouble(value)

56. 将当前字符串转换为Double数值，如果转换失败，则返回指定值，例如指定值配置为1.1d。
    表达式：NumberUtils.toDouble(value,1.1d)

57. 将当前字符串转换为Float数值，如果转换失败，则返回0.0f。
    表达式：NumberUtils.toFloat(value)

58. 将当前字符串转换为Float数值，如果转换失败，则返回指定值，例如配置指定值为1.1f。
    表达式：NumberUtils.toFloat(value,1.1f)

59. 将当前字符串转换为Int数值，如果转换失败，则返回0。
    表达式：NumberUtils.toInt(value)

60. 将当前字符串转换为Int数值，如果转换失败，则返回指定值，例如配置指定值为1。
    表达式：NumberUtils.toInt(value,1)

61. 将字符串转换为Long数值，如果转换失败，则返回0。
    表达式：NumberUtils.toLong(value)

62. 将当前字符串转换为Long数值，如果转换失败，则返回指定值，例如配置指定值为1L。
    表达式：NumberUtils.toLong(value,1L)

63. 将字符串转换为Short数值，如果转换失败，则返回0。
    表达式：NumberUtils.toShort(value)

64. 将当前字符串转换为Short数值，如果转换失败，则返回指定值，例如配置指定值为1。
    表达式：NumberUtils.toShort(value,1)

65. 将当前IP字符串转换为Long数值，例如将“10.78.124.0”转换为LONG数值是“172915712”。
    表达式：CommonUtils.ipToLong(value)

66. 从网络读取一个IP与物理地址映射文件，并存放到Map集合，这里的URL是IP与地址映射文件存放地址，例如“http://10.114.205.45:21203/sqoop/IpList.csv”。
    表达式：HttpsUtils.downloadMap("url")

67. 将IP与地址映射对象缓存起来并指定一个key值用于检索，例如“ipList”。
    表达式：CommonUtils.setCache("ipList",HttpsUtils.downloadMap("url"))

68. 取出缓存的IP与地址映射对象。
    表达式：CommonUtils.getCache("ipList")

69. 判断是否有IP与地址映射缓存。
    表达式：CommonUtils.cacheExists("ipList")

70. 根据指定的偏移类型（month/day/hour/minute/second）及偏移量（正数表示增加，负数表示减少），将指定格式的时间转换为一个新时间，例如将“2019-05-21 12:00:00”增加8个小时。
    表达式：DateUtils.getCurrentTimeByZone("yyyy-MM-dd HH:mm:ss",value, "hour", 8)

Hadoop类型的数据源进行数据迁移时，建议使用的组件版本有哪些？

建议使用的组件版本既可以作为目的端使用，也可以作为源端使用详见下表。

Hadoop类型
组件
说明
MRS/Apache/FusionInsight HD
Hive
暂不支持2.x版本，建议使用的版本：
l    1.2.X
l    3.1.X
HDFS
建议使用的版本：
l    2.8.X
l    3.1.X
Hbase
建议使用的版本：
l    2.1.X
l    1.3.X

数据源为Hive时支持哪些数据格式？

云数据迁移服务支持从Hive数据源读写的数据格式包括SequenceFile、TextFile、ORC、Parquet。

如有疑问，请联系您的客户经理，或致电天翼云官方客服：400-810-9889。

是否支持同步作业到其他集群？

CDM虽然不支持直接在不同集群间迁移作业，但是通过批量导出、批量导入作业的功能，可以间接实现集群间的作业迁移，方法如下：

1. 将CDM集群1中的所有作业批量导出，将作业的JSON文件保存到本地。
   由于安全原因，CDM导出作业时没有导出连接密码，连接密码全部使用“Add password here”替换。

2. 在本地编辑JSON文件，将“Add password here”替换为对应连接的正确密码。

3. 将编辑好的JSON文件批量导入到CDM集群2，实现集群1和集群2之间的作业同步。

是否支持批量创建作业？

CDM可以通过批量导入的功能，实现批量创建作业，方法如下：

1. 手动创建一个作业。

2. 导出作业，将作业的JSON文件保存到本地。

3. 编辑JSON文件，参考该作业的配置，在JSON文件中批量复制出更多作业。

4. 将JSON文件导入CDM集群，实现批量创建作业。

是否支持批量调度作业？

支持。

1. 访问DataArts Studio服务的数据开发模块。

2. 在数据开发主界面的左侧导航栏，选择“数据开发 > 作业开发”，新建作业。

3. 拖动多个CDM Job节点至画布，然后再编排作业。

如何备份CDM作业？

可以，如果用户长时间不需要使用CDM集群，可以将CDM集群停掉或删除来降低成本。

删除前，用户可以先通过CDM的批量导出功能，把所有作业脚本保存到本地，仅在需要的时候再重新创建集群、重新导入作业，实现作业备份。

如果HANA集群只有部分节点和CDM集群网络互通，应该如何配置连接？

如果HANA集群只有部分节点和CDM网络互通，为确保CDM正常连接HANA集群，则需要进行如下配置：

1. 关闭HANA集群的Statement Routing开关。但须注意，关闭Statement Routing，会增加配置节点的压力。

2. 新建HANA连接时，在高级属性中添加属性“distribution”，并将值置为“off”。

完成配置后，CDM即可正常连接HANA集群。

如何使用Java调用CDM的Rest API创建数据迁移作业？

CDM提供了Rest API，可以通过程序调用实现自动化的作业创建或执行控制。

这里以CDM迁移MySQL数据库的表city1的数据到DWS的表city2为例，介绍如何使用Java调用CDM服务的REST API创建、启动、查询、删除该CDM作业。

需要提前准备以下数据：

1. 云账号的用户名、账号名和项目ID。

2. 创建一个CDM集群，并获取集群ID。

   获取方法：在集群管理界面，单击CDM集群名称可查看集群ID，例如“c110beff-0f11-4e75-8b10-da7cd882b0ef”。

3. 创建一个MySQL数据库和一个DWS数据库，并创建好表city1和表city2，创表语句如下：

   MySQL: 
   create table city1(code varchar(10),name varchar(32)); 
   insert into city1 values('NY','New York'); 
   DWS: 
   create table city2(code varchar(10),name varchar(32));

4. 在CDM集群下，创建连接到MySQL的连接，例如连接名称为“mysqltestlink”。创建连接到DWS的连接，例如连接名称为“dwstestlink”。

5. 运行下述代码，依赖HttpClient包，建议使用4.5版本。Maven配置如下：

   <project> 
   <modelVersion>4.0.0</modelVersion> 
   <groupId>cdm</groupId> 
   <artifactId>cdm-client</artifactId> 
   <version>1</version> 
   <dependencies> 
   <dependency> 
   <groupId>org.apache.httpcomponents</groupId> 
   <artifactId>httpclient</artifactId> 
   <version>4.5</version> 
   </dependency> 
   </dependencies> 
   </project>

代码示例

使用Java调用CDM服务的REST API创建、启动、查询、删除CDM作业的代码示例如下：

package cdmclient; 
import java.io.IOException; 
import org.apache.http.Header; 
import org.apache.http.HttpEntity; 
import org.apache.http.HttpHost; 
import org.apache.http.auth.AuthScope; 
import org.apache.http.auth.UsernamePasswordCredentials; 
import org.apache.http.client.CredentialsProvider; 
import org.apache.http.client.config.RequestConfig; 
import org.apache.http.client.methods.CloseableHttpResponse; 
import org.apache.http.client.methods.HttpDelete; 
import org.apache.http.client.methods.HttpGet; 
import org.apache.http.client.methods.HttpPost; 
import org.apache.http.client.methods.HttpPut; 
import org.apache.http.entity.StringEntity; 
import org.apache.http.impl.client.BasicCredentialsProvider; 
import org.apache.http.impl.client.CloseableHttpClient; 
import org.apache.http.impl.client.HttpClients; 
import org.apache.http.util.EntityUtils;
public class CdmClient { 
private final static String DOMAIN_NAME="云账号名"; 
private final static String USER_NAME="云用户名"; 
private final static String USER_PASSWORD="云用户密码"; 
private final static String PROJECT_ID="项目ID"; 
private final static String CLUSTER_ID="CDM集群ID"; 
private final static String JOB_NAME="作业名称"; 
private final static String FROM_LINKNAME="源连接名称"; 
private final static String TO_LINKNAME="目的连接名称"; 
private final static String IAM_ENDPOINT="IAM的Endpoint"; 
private final static String CDM_ENDPOINT="CDM的Endpoint";
private CloseableHttpClient httpclient; 
private String token; 
 
public CdmClient() { 
this.httpclient = createHttpClient(); 
this.token = login(); 
} 
 
private CloseableHttpClient createHttpClient() { 
CloseableHttpClient httpclient =HttpClients.createDefault(); 
return httpclient; 
} 
 
private String login(){ 
HttpPost httpPost = new HttpPost("https://"+IAM_ENDPOINT+"/v3/auth/tokens"); 
String json = 
"{\r\n"+ 
"\"auth\": {\r\n"+ 
"\"identity\": {\r\n"+ 
"\"methods\": [\"password\"],\r\n"+ 
"\"password\": {\r\n"+ 
"\"user\": {\r\n"+ 
"\"name\": \""+USER_NAME+"\",\r\n"+ 
"\"password\": \""+USER_PASSWORD+"\",\r\n"+ 
"\"domain\": {\r\n"+ 
"\"name\": \""+DOMAIN_NAME+"\"\r\n"+ 
"}\r\n"+ 
"}\r\n"+ 
"}\r\n"+ 
"},\r\n"+ 
"\"scope\": {\r\n"+ 
"\"project\": {\r\n"+ 
"\"name\": \"PROJECT_NAME\"\r\n"+ 
"}\r\n"+ 
"}\r\n"+ 
"}\r\n"+ 
"}\r\n"; 
try { 
StringEntity s = new StringEntity(json); 
s.setContentEncoding("UTF-8"); 
s.setContentType("application/json"); 
httpPost.setEntity(s); 
CloseableHttpResponse response = httpclient.execute(httpPost); 
Header tokenHeader = response.getFirstHeader("X-Subject-Token"); 
String token = tokenHeader.getValue(); 
System.out.println("Login successful"); 
return token; 
} catch (Exception e) { 
throw new RuntimeException("login failed.", e); 
} 
}
/*创建作业*/ 
 
public void createJob(){ 
HttpPost httpPost = new HttpPost("https://"+CDM_ENDPOINT+"/cdm/v1.0/"+PROJECT_ID+"/clusters/"+CLUSTER_ID+"/cdm/job"); 
 
/**此处JSON信息比较复杂，可以先在作业管理界面上创建一个作业，然后单击作业后的“作业JSON定义”，复制其中的JSON内容，格式化为Java字符串语法，然后粘贴到此处。 
*JSON消息体中一般只需要替换连接名、导入和导出的表名、导入导出表的字段列表、源表中用于分区的字段。**/ 
 
String json = 
"{\r\n"+ 
"\"jobs\": [\r\n"+ 
"{\r\n"+ 
"\"from-connector-name\": \"generic-jdbc-connector\",\r\n"+ 
"\"name\": \""+JOB_NAME+"\",\r\n"+ 
"\"to-connector-name\": \"generic-jdbc-connector\",\r\n"+ 
"\"driver-config-values\": {\r\n"+ 
"\"configs\": [\r\n"+ 
"{\r\n"+ 
"\"inputs\": [\r\n"+ 
"{\r\n"+ 
"\"name\": \"throttlingConfig.numExtractors\",\r\n"+ 
"\"value\": \"1\"\r\n"+ 
"}\r\n"+ 
"],\r\n"+ 
"\"validators\": [],\r\n"+ 
"\"type\": \"JOB\",\r\n"+ 
"\"id\": 30,\r\n"+ 
"\"name\": \"throttlingConfig\"\r\n"+ 
"}\r\n"+ 
"]\r\n"+ 
"},\r\n"+ 
"\"from-link-name\": \""+FROM_LINKNAME+"\",\r\n"+ 
"\"from-config-values\": {\r\n"+ 
"\"configs\": [\r\n"+ 
"{\r\n"+ 
"\"inputs\": [\r\n"+ 
"{\r\n"+ 
"\"name\": \"fromJobConfig.schemaName\",\r\n"+ 
"\"value\": \"sqoop\"\r\n"+ 
"},\r\n"+ 
"{\r\n"+ 
"\"name\": \"fromJobConfig.tableName\",\r\n"+ 
"\"value\": \"city1\"\r\n"+ 
"},\r\n"+ 
"{\r\n"+ 
"\"name\": \"fromJobConfig.columnList\",\r\n"+ 
"\"value\": \"code&name\"\r\n"+ 
"},\r\n"+ 
"{\r\n"+ 
"\"name\": \"fromJobConfig.partitionColumn\",\r\n"+ 
"\"value\": \"code\"\r\n"+ 
"}\r\n"+ 
"],\r\n"+ 
"\"validators\": [],\r\n"+ 
"\"type\": \"JOB\",\r\n"+ 
"\"id\": 7,\r\n"+ 
"\"name\": \"fromJobConfig\"\r\n"+ 
"}\r\n"+ 
"]\r\n"+ 
"},\r\n"+ 
"\"to-link-name\": \""+TO_LINKNAME+"\",\r\n"+ 
"\"to-config-values\": {\r\n"+ 
"\"configs\": [\r\n"+ 
"{\r\n"+ 
"\"inputs\": [\r\n"+ 
"{\r\n"+ 
"\"name\": \"toJobConfig.schemaName\",\r\n"+ 
"\"value\": \"sqoop\"\r\n"+ 
"},\r\n"+ 
"{\r\n"+ 
"\"name\": \"toJobConfig.tableName\",\r\n"+ 
"\"value\": \"city2\"\r\n"+ 
"},\r\n"+ 
"{\r\n"+ 
"\"name\": \"toJobConfig.columnList\",\r\n"+ 
"\"value\": \"code&name\"\r\n"+ 
"}, \r\n"+ 
"{\r\n"+ 
"\"name\": \"toJobConfig.shouldClearTable\",\r\n"+ 
"\"value\": \"true\"\r\n"+ 
"}\r\n"+ 
"],\r\n"+ 
"\"validators\": [],\r\n"+ 
"\"type\": \"JOB\",\r\n"+ 
"\"id\": 9,\r\n"+ 
"\"name\": \"toJobConfig\"\r\n"+ 
"}\r\n"+ 
"]\r\n"+ 
"}\r\n"+ 
"}\r\n"+ 
"]\r\n"+ 
"}\r\n"; 
try { 
StringEntity s = new StringEntity(json); 
s.setContentEncoding("UTF-8"); 
s.setContentType("application/json"); 
httpPost.setEntity(s); 
httpPost.addHeader("X-Auth-Token", this.token); 
httpPost.addHeader("X-Language", "en-us"); 
CloseableHttpResponse response = httpclient.execute(httpPost); 
int status = response.getStatusLine().getStatusCode(); 
if(status == 200){ 
System.out.println("Create job successful."); 
}else{ 
System.out.println("Create job failed."); 
HttpEntity entity = response.getEntity(); 
System.out.println(EntityUtils.toString(entity)); 
} 
} catch (Exception e) { 
e.printStackTrace(); 
throw new RuntimeException("Create job failed.", e); 
} 
}
/*启动作业*/ 
 
public void startJob(){ 
HttpPut httpPut = new HttpPut("https://"+CDM_ENDPOINT+"/cdm/v1.0/"+PROJECT_ID+"/clusters/"+CLUSTER_ID+"/cdm/job/"+JOB_NAME+"/start"); 
String json = ""; 
try { 
StringEntity s = new StringEntity(json); 
s.setContentEncoding("UTF-8"); 
s.setContentType("application/json"); 
httpPut.setEntity(s); 
httpPut.addHeader("X-Auth-Token", this.token); 
httpPut.addHeader("X-Language", "en-us"); 
CloseableHttpResponse response = httpclient.execute(httpPut); 
int status = response.getStatusLine().getStatusCode(); 
if(status == 200){ 
System.out.println("Start job successful."); 
}else{ 
System.out.println("Start job failed."); 
HttpEntity entity = response.getEntity(); 
System.out.println(EntityUtils.toString(entity)); 
} 
} catch (Exception e) { 
e.printStackTrace(); 
throw new RuntimeException("Start job failed.", e); 
} 
}
/*循环查询作业运行状态，直到作业运行结束。*/ 
 
public void getJobStatus(){ 
HttpGet httpGet = new HttpGet("https://"+CDM_ENDPOINT+"/cdm/v1.0/"+PROJECT_ID+"/clusters/"+CLUSTER_ID+"/cdm/job/"+JOB_NAME+"/status"); 
try { 
httpGet.addHeader("X-Auth-Token", this.token); 
httpGet.addHeader("X-Language", "en-us"); 
boolean flag = true; 
while(flag){ 
CloseableHttpResponse response = httpclient.execute(httpGet); 
int status = response.getStatusLine().getStatusCode(); 
if(status == 200){ 
HttpEntity entity = response.getEntity(); 
String msg = EntityUtils.toString(entity); 
if(msg.contains("\"status\":\"SUCCEEDED\"")){ 
System.out.println("Job succeeded"); 
break; 
}else if (msg.contains("\"status\":\"FAILED\"")){ 
System.out.println("Job failed."); 
break; 
}else{ 
Thread.sleep(1000); 
} 
 
}else{ 
System.out.println("Get job status failed."); 
HttpEntity entity = response.getEntity(); 
System.out.println(EntityUtils.toString(entity)); 
break; 
} 
} 
} catch (Exception e) { 
e.printStackTrace(); 
throw new RuntimeException("Get job status failed.", e); 
} 
} 

/*删除作业*/ 
 
public void deleteJob(){ 
HttpDelete httpDelte = new HttpDelete("https://"+CDM_ENDPOINT+"/cdm/v1.0/"+PROJECT_ID+"/clusters/"+CLUSTER_ID+"/cdm/job/"+JOB_NAME); 
try { 
httpDelte.addHeader("X-Auth-Token", this.token); 
httpDelte.addHeader("X-Language", "en-us"); 
CloseableHttpResponse response = httpclient.execute(httpDelte); 
int status = response.getStatusLine().getStatusCode(); 
if(status == 200){ 
System.out.println("Delete job successful."); 
}else{ 
System.out.println("Delete job failed."); 
HttpEntity entity = response.getEntity(); 
System.out.println(EntityUtils.toString(entity)); 
} 
} catch (Exception e) { 
e.printStackTrace(); 
throw new RuntimeException("Delete job failed.", e); 
} 
}
/*关闭*/ 
 
public void close(){ 
try { 
httpclient.close(); 
} catch (IOException e) { 
throw new RuntimeException("Close failed.", e); 
} 
} 
 
public static void main(String[] args){ 
CdmClient cdmClient = new CdmClient(); 
cdmClient.createJob(); 
cdmClient.startJob(); 
cdmClient.getJobStatus(); 
cdmClient.deleteJob(); 
cdmClient.close(); 
} 
}

如何将云下内网或第三方云上的私网与CDM连通？

很多企业会把关键数据源建设在内网，例如数据库、文件服务器等。由于CDM运行在云上，如果要通过CDM迁移内网数据到云上的话，可以通过以下几种方式连通内网和CDM的网络：

• 如果目标数据源为云下的数据库，则需要通过公网或者专线打通网络。通过公网互通时，需确保CDM集群已绑定EIP、CDM云上安全组出方向放通云下数据源所在的主机、数据源所在的主机可以访问公网且防火墙规则已开放连接端口。

• 在本地数据中心和云服务VPC之间建立VPN通道。

• 通过NAT（网络地址转换，Network

  Address Translation）或端口转发，以代理的方式访问。

这里重点介绍如何通过端口转发工具来实现访问内部数据，流程如下：

1. 找一台windows机器作为网关，该机器必须可以直接访问Internet，同时可以访问内网。

2. 在该机器上安装端口映射工具（IPOP）。

3. 通过端口映射工具（IPOP）配置端口映射。

说明长时间将内网数据库暴露在公网会有安全风险，迁移数据完成后，请及时停止端口映射。

场景描述

这里假设是将内网MySQL迁移到云服务DWS。

图中的内网既可以是企业自己的数据中心，也可以是在第三方云的虚拟数据中心私网。

如下图“网络拓扑样例”：

操作步骤

1.找一台Windows机器作为网关机，该机器同时配置内网和外网IP。通过以下测试来确保网关机器的服务要求：

a.在该机器上ping内网MySQL地址可以ping通，例如：ping 192.168.1.8。

b.在另外一台可上网的机器上ping网关机的公网地址可以ping通，例如ping 202.xx.xx.10。

2.下载端口映射工具IPOP，在网关机上安装IPOP。

3.运行端口映射工具，选择“端口映射”，如图下图配置端口映射所示。

• 本地地址、本地端口：配置为网关机的公网地址和端口（后续在CDM上创建MySQL连接时输入这个地址和端口）。

• 映射地址、映射端口：配置为内网MySQL的地址和端口。
  

4.单击“增加”，添加端口映射关系。

5.单击“开始映射”，这时才会真正开始映射，接收数据包。

至此，就可以在CDM上通过弹性IP读取本地内网MySQL的数据，然后导入到云服务DWS中。

说明
1.    CDM要访问本地数据源，也必须给CDM集群配置EIP。
2.    一般云服务DWS默认也是只允许VPC内部访问，创建CDM集群时，必须将CDM的VPC与DWS配置一致，且推荐在同一个内网和安全组，如果不同，还需要配置允许两个安全组之间的数据访问。
3.    端口映射不仅可以用于迁移内网数据库的数据，还可以迁移例如SFTP服务器上的数据。
4.    Linux机器也可以通过IPTABLE实现端口映射。
5.    内网中的FTP通过端口映射到公网时，需要检查是否启用了PASV模式。这种情况下客户端和服务端建立连接的时候是走的随机端口，所以除了配置21端口映射外，还需要配置PASV模式的端口范围映射，例如vsftp通过配置pasv_min_port和pasv_max_port指定端口范围。

CDM迁移作业的抽取并发数应该如何设置？

CDM迁移作业的抽取并发数，与集群规格和表大小有关。并发抽取数取值范围为1-300，若配置过大，则以队列的形式进行排队。

建议每1CUs（1CUs=1核4G）配置为4，如下表“抽取并发数参考配置”所示，您也可以根据实际情况进行调整。另外，每行数据大小为1MB以下的可以多并发抽取，超过1MB的建议单线程抽取数据。

说明l  迁移的目的端为文件时，CDM不支持多并发，此时应配置为单进程抽取数据。l  单作业的抽取并发数，受到作业“配置管理”中所配置的“最大抽取并发数”影响。“最大抽取并发数”配置的是抽取并发总数。

CDM是否支持动态数据实时迁移功能？

不支持。

如果源端在迁移过程中写数据，可能会出现报错。

故障处理类

问题描述

使用CDM从OBS导入数据到SQL Server时，作业运行失败，错误提示为：Unable to execute the SQL statement. Cause : 将截断字符串或二进制数据。

原因分析

用户OBS中的数据超出了SQL Server数据库的字段长度限制。

解决方法

在SQL Server数据库中建表时，将数据库字段改大，长度不能小于源端OBS中的数据长度。

Oracle迁移到DWS报错ORA-01555

问题现象

使用CDM迁移Oracle数据至DWS，报错图下图所示。

原因分析

1. 数据迁移，整表查询且该表数据量大，那么查询时间较长。

2. 查询过程中，其他用户频繁进行commit操作。

3. Oracel的RBS(rollbackspace 回滚时使用的表空间 )较小，造成迁移任务没有完成，源库已更新，回滚超时。

建议与总结

1. 调小每次查询的数据量。

2. 通过修改数据库配置调大Oracle的RBS。

MongoDB连接迁移失败时如何处理？

在默认情况下，userAdmin角色只具备对角色和用户的管理，不具备对库的读和写权限。

当用户选择MongoDB连接迁移失败时，用户需查看MongoDB连接中用户的权限信息，确保对指定库具备ReadWrite权限。

Hive迁移作业长时间卡住怎么办？

为避免Hive迁移作业长时间卡住，可手动停止迁移作业后，通过编辑Hive连接增加如下属性设置：

• 属性名称：hive.server2.idle.operation.timeout

• 值：10m

如下图所示

使用CDM迁移数据由于字段类型映射不匹配导致报错怎么处理？

问题描述

MySQL迁移时报错：Unable to connect to the database server. Cause: connect timed out.

原因分析

这种情况是由于表数据量较大，并且源端通过where语句过滤，但并非索引列，或列值不离散，查询会全表扫描，导致JDBC连接超时。例如图“非索引列”所示c_date字段为非索引列。

解决方案

1. 优先联系DBA修改表结构，将需要过滤的列配置为索引列，然后重试。
   如果由于数据不离散，导致还是失败请参考2~4，通过增大JDBC超时时间解决。

2. 根据作业找到对应的MySQL连接名称，查找连接信息。
   如下图图连接信息

   

3. 单击“连接管理”，在“操作”列中，单击“连接”进行编辑。

   如下图连接

   

4. 打开高级属性，在“连接属性”中建议新增“connectTimeout”与“socketTimeout”参数及参数值，单击“保存”。

   如下图编辑高级属性：

   

创建了Hive到DWS类型的连接，进行CDM传输任务失败时如何处理？

建议清空历史数据后再次尝试该任务。在使用CDM迁移作业的时候需要配置清空历史数据，然后再做迁移，可大大降低任务失败的概率。

如何使用CDM服务将MySQL的数据导出成SQL文件，然后上传到OBS桶?

CDM服务暂不支持该操作。

建议通过手动导出MySQL的数据文件，然后在服务器上开启SFTP服务，然后新建CDM作业，源端是SFTP协议，目的端是OBS，将文件传过去。

如何处理CDM从OBS迁移数据到DLI出现迁移中断失败的问题？

此类作业问题表现为配置了脏数据写入，但并无脏数据。

这种情况下需要调低并发任务数，即可避免此类问题。

如何处理CDM连接器报错“配置项 [linkConfig.iamAuth] 不存在”？

此问题归因为：客户证书过期。

客户证书过期，需要完成更新证书操作，完成后重新配置连接器即可。

创建数据连接时报错“配置项[linkConfig.createBackendLinks]不存在”或创建作业时报错“配置项 [throttlingConfig.concurrentSubJobs] 不存在”怎么办?

当同时存在多个不同版本的集群，先在低版本CDM集群创建数据连接或保存作业时后，再进入高版本CDM集群时，会偶现此类故障。

需手动清理浏览器缓存，即可避免此类问题。

新建MRS Hive连接时，提示：CORE_0031:Connect time out. (Cdm.0523) 怎么解决？

新建MRS Hive连接时，提示无法下载配置文件，实际是用户权限不足。建议您新建一个业务用户，给对应的权限后重试即可。

如果要创建MRS安全集群的数据连接，不能使用admin用户。因为admin用户是默认的管理页面用户，这个用户无法作为安全集群的认证用户来使用。您可以创建一个新的MRS用户，然后在创建MRS数据连接时，“用户名”和“密码”填写为新建的MRS用户及其密码。

说明l  如果CDM集群为2.9.0版本及之后版本，且MRS集群为3.1.0及之后版本，则所创建的用户至少需具备Manager_viewer的角色权限才能在CDM创建连接；如果需要对MRS组件的库、表、列进行操作，还需要参考MRS文档添加对应组件的库、表、列操作权限。l  如果CDM集群为2.9.0之前的版本，或MRS集群为3.1.0之前的版本，则所创建的用户需要具备Manager_administrator或System_administrator权限，才能在CDM创建连接。l  仅具备Manager_tenant或Manager_auditor权限，无法创建连接。

迁移时已选择表不存在时自动创表，提示“CDM not support auto create empty table with no column”怎么处理？

这是由于数据库表名中含有特殊字符导致识别出语法错误，按数据库对象命名规则重新命名后恢复正常。

例如，DWS数据仓库中的数据表命名需要满足以下约束：长度不超过63个字符，以字母或下划线开头，中间字符可以是字母、数字、下划线、$、#。

创建Oracle关系型数据库迁移作业时，无法获取模式名怎么处理？

这是由于可能上传了暂不支持的最新ORACLE_8驱动（如Oracle Database 21c (21.3) drivers）。

解决此问题，推荐使用Oracle Database 12c中的ojdbc8.jar驱动。