wyl/DataX
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

DataX 通道能力更新（MaxCompute、Hologres、Tdengine、OSS）、安全漏洞更新、通用打包更新等

Browse Source
This commit is contained in:
jt-chen
2022-06-09 21:13:49 +08:00
parent 3225f02bf9
commit e9a579759e
261 changed files with 17980 additions and 2432 deletions
Download Patch File Download Diff File Expand all files Collapse all files
odpsreader/pom.xml
+23 -30
View File
@@ -36,18 +36,18 @@
<artifactId>guava</artifactId>
<version>16.0.1</version>
</dependency>
<dependency>
<groupId>org.bouncycastle</groupId>
<artifactId>bcprov-jdk15on</artifactId>
<version>1.52</version>
<scope>system</scope>
<systemPath>${basedir}/src/main/libs/bcprov-jdk15on-1.52.jar</systemPath>
</dependency>
<dependency>
<groupId>com.aliyun.odps</groupId>
<artifactId>odps-sdk-core</artifactId>
<version>0.20.7-public</version>
</dependency>
<dependency>
<groupId>org.xerial</groupId>
<artifactId>sqlite-jdbc</artifactId>
<version>3.34.0</version>
</dependency>
<!-- ref:http://odps.alibaba-inc.com/doc/prddoc/odps_sdk_v2/sdk.html -->
<dependency>
<groupId>com.aliyun.odps</groupId>
<artifactId>odps-sdk-core</artifactId>
<version>0.38.4-public</version>
</dependency>
<dependency>
<groupId>org.mockito</groupId>
@@ -87,29 +87,22 @@
<version>1.4.10</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.mockito</groupId>
<artifactId>mockito-core</artifactId>
<version>1.8.5</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.powermock</groupId>
<artifactId>powermock-api-mockito</artifactId>
<version>1.4.10</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.powermock</groupId>
<artifactId>powermock-module-junit4</artifactId>
<version>1.4.10</version>
<scope>test</scope>
<groupId>commons-codec</groupId>
<artifactId>commons-codec</artifactId>
<version>1.8</version>
</dependency>
</dependencies>
<build>
<resources>
<resource>
<directory>src/main/java</directory>
<includes>
<include>**/*.properties</include>
</includes>
</resource>
</resources>
<plugins>
<!-- compiler plugin -->
<plugin>
odpsreader/src/main/assembly/package.xml
-7
View File
@@ -23,13 +23,6 @@
</includes>
<outputDirectory>plugin/reader/odpsreader</outputDirectory>
</fileSet>
<fileSet>
<directory>src/main/libs</directory>
<includes>
<include>*.*</include>
</includes>
<outputDirectory>plugin/reader/odpsreader/libs</outputDirectory>
</fileSet>
</fileSets>
<dependencySets>
odpsreader/src/main/java/com/alibaba/datax/plugin/reader/odpsreader/Constant.java
+3 -1
View File
@@ -31,5 +31,7 @@ public class Constant {
public static final String PARTITION_COLUMNS = "partitionColumns";
public static final String PARSED_COLUMNS = "parsedColumns";
public static final String PARTITION_FILTER_HINT = "/*query*/";
}
odpsreader/src/main/java/com/alibaba/datax/plugin/reader/odpsreader/Key.java
+11
View File
@@ -5,6 +5,8 @@ public class Key {
public final static String ACCESS_ID = "accessId";
public final static String ACCESS_KEY = "accessKey";
public final static String SECURITY_TOKEN = "securityToken";
public static final String PROJECT = "project";
@@ -30,5 +32,14 @@ public class Key {
public final static String IS_COMPRESS = "isCompress";
public final static String MAX_RETRY_TIME = "maxRetryTime";
// 分区不存在时
public final static String SUCCESS_ON_NO_PATITION="successOnNoPartition";
// preSql
public final static String PRE_SQL="preSql";
// postSql
public final static String POST_SQL="postSql";
}
odpsreader/src/main/java/com/alibaba/datax/plugin/reader/odpsreader/LocalStrings.properties
+64
View File
@@ -0,0 +1,64 @@
description.DATAX_R_ODPS_001=\u7F3A\u5C11\u5FC5\u586B\u53C2\u6570
description.DATAX_R_ODPS_002=\u914D\u7F6E\u503C\u4E0D\u5408\u6CD5
description.DATAX_R_ODPS_003=\u521B\u5EFAODPS Session\u5931\u8D25
description.DATAX_R_ODPS_004=\u83B7\u53D6ODPS Session\u5931\u8D25
description.DATAX_R_ODPS_005=\u8BFB\u53D6ODPS\u6570\u636E\u5931\u8D25
description.DATAX_R_ODPS_006=\u83B7\u53D6AK\u5931\u8D25
description.DATAX_R_ODPS_007=\u8BFB\u53D6\u6570\u636E\u53D1\u751F\u5F02\u5E38
description.DATAX_R_ODPS_008=\u6253\u5F00RecordReader\u5931\u8D25
description.DATAX_R_ODPS_009=ODPS\u9879\u76EE\u4E0D\u5B58\u5728
description.DATAX_R_ODPS_010=\u8868\u4E0D\u5B58\u5728
description.DATAX_R_ODPS_011=AK\u4E0D\u5B58\u5728
description.DATAX_R_ODPS_012=AK\u975E\u6CD5
description.DATAX_R_ODPS_013=AK\u62D2\u7EDD\u8BBF\u95EE
description.DATAX_R_ODPS_014=splitMode\u914D\u7F6E\u9519\u8BEF
description.DATAX_R_ODPS_015=ODPS\u8D26\u53F7\u7C7B\u578B\u9519\u8BEF
description.DATAX_R_ODPS_016=\u4E0D\u652F\u6301\u89C6\u56FE
description.DATAX_R_ODPS_017=\u5206\u533A\u914D\u7F6E\u9519\u8BEF
description.DATAX_R_ODPS_018=\u5206\u533A\u4E0D\u5B58\u5728
description.DATAX_R_ODPS_019=\u6267\u884CODPS SQL\u5931\u8D25
description.DATAX_R_ODPS_020=\u6267\u884CODPS SQL\u53D1\u751F\u5F02\u5E38
solution.DATAX_R_ODPS_001=\u8BF7\u4FEE\u6539\u914D\u7F6E\u6587\u4EF6
solution.DATAX_R_ODPS_002=\u8BF7\u4FEE\u6539\u914D\u7F6E\u503C
solution.DATAX_R_ODPS_003=\u8BF7\u786E\u5B9A\u914D\u7F6E\u7684AK\u6216\u8054\u7CFBODPS\u7BA1\u7406\u5458
solution.DATAX_R_ODPS_004=\u8BF7\u8054\u7CFBODPS\u7BA1\u7406\u5458
solution.DATAX_R_ODPS_005=\u8BF7\u8054\u7CFBODPS\u7BA1\u7406\u5458
solution.DATAX_R_ODPS_006=\u8BF7\u786E\u5B9A\u914D\u7F6E\u7684AK
solution.DATAX_R_ODPS_007=\u8BF7\u8054\u7CFBODPS\u7BA1\u7406\u5458
solution.DATAX_R_ODPS_008=\u8BF7\u8054\u7CFBODPS\u7BA1\u7406\u5458
solution.DATAX_R_ODPS_009=\u8BF7\u786E\u5B9A\u914D\u7F6E\u7684\u9879\u76EE\u540D
solution.DATAX_R_ODPS_010=\u8BF7\u786E\u5B9A\u914D\u7F6E\u7684\u8868\u540D
solution.DATAX_R_ODPS_011=\u8BF7\u786E\u5B9A\u914D\u7F6E\u7684AK
solution.DATAX_R_ODPS_012=\u8BF7\u4FEE\u6539AK
solution.DATAX_R_ODPS_013=\u8BF7\u786E\u5B9AAK\u5728\u9879\u76EE\u4E2D\u7684\u6743\u9650
solution.DATAX_R_ODPS_014=\u8BF7\u4FEE\u6539splitMode\u503C
solution.DATAX_R_ODPS_015=\u8BF7\u4FEE\u6539\u8D26\u53F7\u7C7B\u578B
solution.DATAX_R_ODPS_016=\u8BF7\u4FEE\u6539\u914D\u7F6E\u6587\u4EF6
solution.DATAX_R_ODPS_017=\u8BF7\u4FEE\u6539\u5206\u533A\u503C
solution.DATAX_R_ODPS_018=\u8BF7\u4FEE\u6539\u914D\u7F6E\u7684\u5206\u533A\u503C
solution.DATAX_R_ODPS_019=\u8BF7\u8054\u7CFBODPS\u7BA1\u7406\u5458
solution.DATAX_R_ODPS_020=\u8BF7\u8054\u7CFBODPS\u7BA1\u7406\u5458
odpsreader.1=\u6E90\u5934\u8868:{0} \u662F\u865A\u62DF\u89C6\u56FE\uFF0CDataX \u4E0D\u652F\u6301\u8BFB\u53D6\u865A\u62DF\u89C6\u56FE.
odpsreader.2=\u60A8\u6240\u914D\u7F6E\u7684 splitMode:{0} \u4E0D\u6B63\u786E. splitMode \u4EC5\u5141\u8BB8\u914D\u7F6E\u4E3A record \u6216\u8005 partition.
odpsreader.3=\u5206\u533A\u4FE1\u606F\u6CA1\u6709\u914D\u7F6E.\u7531\u4E8E\u6E90\u5934\u8868:{0} \u4E3A\u5206\u533A\u8868, \u6240\u4EE5\u60A8\u9700\u8981\u914D\u7F6E\u5176\u62BD\u53D6\u7684\u8868\u7684\u5206\u533A\u4FE1\u606F. \u683C\u5F0F\u5F62\u5982:pt=hello,ds=hangzhou\uFF0C\u8BF7\u60A8\u53C2\u8003\u6B64\u683C\u5F0F\u4FEE\u6539\u8BE5\u914D\u7F6E\u9879.
odpsreader.4=\u5206\u533A\u4FE1\u606F\u914D\u7F6E\u9519\u8BEF.\u6E90\u5934\u8868:{0} \u867D\u7136\u4E3A\u5206\u533A\u8868, \u4F46\u5176\u5B9E\u9645\u5206\u533A\u503C\u5E76\u4E0D\u5B58\u5728. \u8BF7\u786E\u8BA4\u6E90\u5934\u8868\u5DF2\u7ECF\u751F\u6210\u8BE5\u5206\u533A\uFF0C\u518D\u8FDB\u884C\u6570\u636E\u62BD\u53D6.
odpsreader.5=\u5206\u533A\u914D\u7F6E\u9519\u8BEF\uFF0C\u6839\u636E\u60A8\u6240\u914D\u7F6E\u7684\u5206\u533A\u6CA1\u6709\u5339\u914D\u5230\u6E90\u5934\u8868\u4E2D\u7684\u5206\u533A. \u6E90\u5934\u8868\u6240\u6709\u5206\u533A\u662F:[\n{0}\n], \u60A8\u914D\u7F6E\u7684\u5206\u533A\u662F:[\n{1}\n]. \u8BF7\u60A8\u6839\u636E\u5B9E\u9645\u60C5\u51B5\u518D\u4F5C\u51FA\u4FEE\u6539.
odpsreader.6=\u5206\u533A\u914D\u7F6E\u9519\u8BEF\uFF0C\u6E90\u5934\u8868:{0} \u4E3A\u975E\u5206\u533A\u8868, \u60A8\u4E0D\u80FD\u914D\u7F6E\u5206\u533A. \u8BF7\u60A8\u5220\u9664\u8BE5\u914D\u7F6E\u9879.
odpsreader.7=\u6E90\u5934\u8868:{0} \u7684\u6240\u6709\u5206\u533A\u5217\u662F:[{1}]
odpsreader.8=\u5206\u533A\u914D\u7F6E\u9519\u8BEF, \u60A8\u6240\u914D\u7F6E\u7684\u5206\u533A\u7EA7\u6570\u548C\u8BE5\u8868\u7684\u5B9E\u9645\u60C5\u51B5\u4E0D\u4E00\u81F4, \u6BD4\u5982\u5206\u533A:[{0}] \u662F {1} \u7EA7\u5206\u533A, \u800C\u5206\u533A:[{2}] \u662F {3} \u7EA7\u5206\u533A. DataX \u662F\u901A\u8FC7\u82F1\u6587\u9017\u53F7\u5224\u65AD\u60A8\u6240\u914D\u7F6E\u7684\u5206\u533A\u7EA7\u6570\u7684. \u6B63\u786E\u7684\u683C\u5F0F\u5F62\u5982\"pt=$'{bizdate'}, type=0\" \uFF0C\u8BF7\u60A8\u53C2\u8003\u793A\u4F8B\u4FEE\u6539\u8BE5\u914D\u7F6E\u9879.
odpsreader.9=\u5206\u533A\u914D\u7F6E\u9519\u8BEF, \u60A8\u6240\u914D\u7F6E\u7684\u5206\u533A:{0} \u7684\u7EA7\u6570:{1} \u4E0E\u60A8\u8981\u8BFB\u53D6\u7684 ODPS \u6E90\u5934\u8868\u7684\u5206\u533A\u7EA7\u6570:{2} \u4E0D\u76F8\u7B49. DataX \u662F\u901A\u8FC7\u82F1\u6587\u9017\u53F7\u5224\u65AD\u60A8\u6240\u914D\u7F6E\u7684\u5206\u533A\u7EA7\u6570\u7684.\u6B63\u786E\u7684\u683C\u5F0F\u5F62\u5982\"pt=$'{bizdate'}, type=0\" \uFF0C\u8BF7\u60A8\u53C2\u8003\u793A\u4F8B\u4FEE\u6539\u8BE5\u914D\u7F6E\u9879.
odpsreader.10=\u6E90\u5934\u8868:{0} \u7684\u6240\u6709\u5B57\u6BB5\u662F:[{1}]
odpsreader.11=\u8FD9\u662F\u4E00\u6761\u8B66\u544A\u4FE1\u606F\uFF0C\u60A8\u914D\u7F6E\u7684 ODPS \u8BFB\u53D6\u7684\u5217\u4E3A*\uFF0C\u8FD9\u662F\u4E0D\u63A8\u8350\u7684\u884C\u4E3A\uFF0C\u56E0\u4E3A\u5F53\u60A8\u7684\u8868\u5B57\u6BB5\u4E2A\u6570\u3001\u7C7B\u578B\u6709\u53D8\u52A8\u65F6\uFF0C\u53EF\u80FD\u5F71\u54CD\u4EFB\u52A1\u6B63\u786E\u6027\u751A\u81F3\u4F1A\u8FD0\u884C\u51FA\u9519. \u5EFA\u8BAE\u60A8\u628A\u6240\u6709\u9700\u8981\u62BD\u53D6\u7684\u5217\u90FD\u914D\u7F6E\u4E0A.
odpsreader.12=\u6E90\u5934\u8868:{0} \u7684\u5206\u533A:{1} \u6CA1\u6709\u5185\u5BB9\u53EF\u62BD\u53D6, \u8BF7\u60A8\u77E5\u6653.
odpsreader.13=\u6E90\u5934\u8868:{0} \u7684\u5206\u533A:{1} \u8BFB\u53D6\u884C\u6570\u4E3A\u8D1F\u6570, \u8BF7\u8054\u7CFB ODPS \u7BA1\u7406\u5458\u67E5\u770B\u8868\u72B6\u6001!
odpsreader.14=\u6E90\u5934\u8868:{0} \u7684\u5206\u533A:{1} \u8BFB\u53D6\u5931\u8D25, \u8BF7\u8054\u7CFB ODPS \u7BA1\u7406\u5458\u67E5\u770B\u9519\u8BEF\u8BE6\u60C5.
readerproxy.1=odps-read-exception, \u91CD\u8BD5\u7B2C{0}\u6B21
readerproxy.2=\u60A8\u7684\u5206\u533A [{0}] \u89E3\u6790\u51FA\u73B0\u9519\u8BEF,\u89E3\u6790\u540E\u6B63\u786E\u7684\u914D\u7F6E\u65B9\u5F0F\u7C7B\u4F3C\u4E3A [ pt=1,dt=1 ].
readerproxy.3=\u8868\u6240\u6709\u5206\u533A\u4FE1\u606F\u4E3A: {0} \u5176\u4E2D\u627E\u4E0D\u5230 [{1}] \u5BF9\u5E94\u7684\u5206\u533A\u503C.
readerproxy.4=\u60A8\u8BFB\u53D6\u5206\u533A [{0}] \u51FA\u73B0\u65E5\u671F\u8F6C\u6362\u5F02\u5E38, \u65E5\u671F\u7684\u5B57\u7B26\u4E32\u8868\u793A\u4E3A [{1}].
readerproxy.5=DataX \u62BD\u53D6 ODPS \u6570\u636E\u4E0D\u652F\u6301\u5B57\u6BB5\u7C7B\u578B\u4E3A:[{0}]. \u76EE\u524D\u652F\u6301\u62BD\u53D6\u7684\u5B57\u6BB5\u7C7B\u578B\u6709\uFF1Abigint, boolean, datetime, double, decimal, string. \u60A8\u53EF\u4EE5\u9009\u62E9\u4E0D\u62BD\u53D6 DataX \u4E0D\u652F\u6301\u7684\u5B57\u6BB5\u6216\u8005\u8054\u7CFB ODPS \u7BA1\u7406\u5458\u5BFB\u6C42\u5E2E\u52A9.
odpsreader/src/main/java/com/alibaba/datax/plugin/reader/odpsreader/OdpsReader.java
+243 -97
View File
@@ -5,44 +5,44 @@ import com.alibaba.datax.common.plugin.RecordSender;
import com.alibaba.datax.common.spi.Reader;
import com.alibaba.datax.common.util.Configuration;
import com.alibaba.datax.common.util.FilterUtil;
import com.alibaba.datax.plugin.reader.odpsreader.util.IdAndKeyUtil;
import com.alibaba.datax.plugin.reader.odpsreader.util.OdpsSplitUtil;
import com.alibaba.datax.plugin.reader.odpsreader.util.OdpsUtil;
import com.aliyun.odps.*;
import com.alibaba.datax.common.util.MessageSource;
import com.alibaba.datax.plugin.reader.odpsreader.util.*;
import com.alibaba.fastjson.JSON;
import com.aliyun.odps.Column;
import com.aliyun.odps.Odps;
import com.aliyun.odps.Table;
import com.aliyun.odps.TableSchema;
import com.aliyun.odps.tunnel.TableTunnel.DownloadSession;
import com.aliyun.odps.type.TypeInfo;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.tuple.MutablePair;
import org.apache.commons.lang3.tuple.Pair;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.*;
public class OdpsReader extends Reader {
public static class Job extends Reader.Job {
private static final Logger LOG = LoggerFactory
.getLogger(Job.class);
private static boolean IS_DEBUG = LOG.isDebugEnabled();
.getLogger(Job.class);
private static final MessageSource MESSAGE_SOURCE = MessageSource.loadResourceBundle(OdpsReaderErrorCode.class, Locale.ENGLISH, MessageSource.timeZone);
private Configuration originalConfig;
private boolean successOnNoPartition;
private Odps odps;
private Table table;
@Override
public void preCheck() {
this.init();
this.prepare();
}
@Override
public void init() {
this.originalConfig = super.getPluginJobConf();
this.successOnNoPartition = this.originalConfig.getBool(Key.SUCCESS_ON_NO_PATITION, false);
//如果用户没有配置accessId/accessKey,尝试从环境变量获取
String accountType = originalConfig.getString(Key.ACCOUNT_TYPE, Constant.DEFAULT_ACCOUNT_TYPE);
@@ -59,17 +59,21 @@ public class OdpsReader extends Reader {
dealSplitMode(this.originalConfig);
this.odps = OdpsUtil.initOdps(this.originalConfig);
}
private void initOdpsTableInfo() {
String tableName = this.originalConfig.getString(Key.TABLE);
String projectName = this.originalConfig.getString(Key.PROJECT);
this.table = OdpsUtil.getTable(this.odps, projectName, tableName);
this.originalConfig.set(Constant.IS_PARTITIONED_TABLE,
OdpsUtil.isPartitionedTable(table));
OdpsUtil.isPartitionedTable(table));
boolean isVirtualView = this.table.isVirtualView();
if (isVirtualView) {
throw DataXException.asDataXException(OdpsReaderErrorCode.VIRTUAL_VIEW_NOT_SUPPORT,
String.format("源头表:%s 是虚拟视图，DataX 不支持读取虚拟视图.", tableName));
MESSAGE_SOURCE.message("odpsreader.1", tableName));
}
this.dealPartition(this.table);
@@ -79,11 +83,11 @@ public class OdpsReader extends Reader {
private void dealSplitMode(Configuration originalConfig) {
String splitMode = originalConfig.getString(Key.SPLIT_MODE, Constant.DEFAULT_SPLIT_MODE).trim();
if (splitMode.equalsIgnoreCase(Constant.DEFAULT_SPLIT_MODE) ||
splitMode.equalsIgnoreCase(Constant.PARTITION_SPLIT_MODE)) {
splitMode.equalsIgnoreCase(Constant.PARTITION_SPLIT_MODE)) {
originalConfig.set(Key.SPLIT_MODE, splitMode);
} else {
throw DataXException.asDataXException(OdpsReaderErrorCode.SPLIT_MODE_ERROR,
String.format("您所配置的 splitMode:%s 不正确. splitMode 仅允许配置为 record 或者 partition.", splitMode));
MESSAGE_SOURCE.message("odpsreader.2", splitMode));
}
}
@@ -98,7 +102,7 @@ public class OdpsReader extends Reader {
*/
private void dealPartition(Table table) {
List<String> userConfiguredPartitions = this.originalConfig.getList(
Key.PARTITION, String.class);
Key.PARTITION, String.class);
boolean isPartitionedTable = this.originalConfig.getBool(Constant.IS_PARTITIONED_TABLE);
List<String> partitionColumns = new ArrayList<String>();
@@ -107,60 +111,140 @@ public class OdpsReader extends Reader {
// 分区表，需要配置分区
if (null == userConfiguredPartitions || userConfiguredPartitions.isEmpty()) {
throw DataXException.asDataXException(OdpsReaderErrorCode.PARTITION_ERROR,
String.format("分区信息没有配置.由于源头表:%s 为分区表, 所以您需要配置其抽取的表的分区信息. 格式形如:pt=hello,ds=hangzhou，请您参考此格式修改该配置项.",
table.getName()));
MESSAGE_SOURCE.message("odpsreader.3", table.getName()));
} else {
List<String> allPartitions = OdpsUtil.getTableAllPartitions(table);
if (null == allPartitions || allPartitions.isEmpty()) {
throw DataXException.asDataXException(OdpsReaderErrorCode.PARTITION_ERROR,
String.format("分区信息配置错误.源头表:%s 虽然为分区表, 但其实际分区值并不存在. 请确认源头表已经生成该分区，再进行数据抽取.",
table.getName()));
}
List<String> parsedPartitions = expandUserConfiguredPartition(
allPartitions, userConfiguredPartitions);
if (null == parsedPartitions || parsedPartitions.isEmpty()) {
throw DataXException.asDataXException(
OdpsReaderErrorCode.PARTITION_ERROR,
String.format(
"分区配置错误，根据您所配置的分区没有匹配到源头表中的分区. 源头表所有分区是:[\n%s\n], 您配置的分区是:[\n%s\n]. 请您根据实际情况在作出修改. ",
StringUtils.join(allPartitions, "\n"),
StringUtils.join(userConfiguredPartitions, "\n")));
}
this.originalConfig.set(Key.PARTITION, parsedPartitions);
for (Column column : table.getSchema()
.getPartitionColumns()) {
// 获取分区列名, 支持用户配置分区列同步
for (Column column : table.getSchema().getPartitionColumns()) {
partitionColumns.add(column.getName());
}
List<String> allPartitions = OdpsUtil.getTableAllPartitions(table);
List<String> parsedPartitions = expandUserConfiguredPartition(
table, allPartitions, userConfiguredPartitions, partitionColumns.size());
if (null == parsedPartitions || parsedPartitions.isEmpty()) {
if (!this.successOnNoPartition) {
// PARTITION_NOT_EXISTS_ERROR 这个异常ErrorCode在AdsWriter有使用，用户判断空分区Load Data任务不报错
// 其他类型的异常不要使用这个错误码
throw DataXException.asDataXException(
OdpsReaderErrorCode.PARTITION_NOT_EXISTS_ERROR,
MESSAGE_SOURCE.message("odpsreader.5",
StringUtils.join(allPartitions, "\n"),
StringUtils.join(userConfiguredPartitions, "\n")));
} else {
LOG.warn(
String.format(
"The partition configuration is wrong, " +
"but you have configured the successOnNoPartition to be true to ignore the error. " +
"According to the partition you have configured, it does not match the partition in the source table. " +
"All the partitions in the source table are:[\n%s\n], the partition you configured is:[\n%s\n]. " +
"please revise it according to the actual situation.",
StringUtils.join(allPartitions, "\n"),
StringUtils.join(userConfiguredPartitions, "\n")));
}
}
LOG.info(String
.format("expand user configured partitions are : %s", JSON.toJSONString(parsedPartitions)));
this.originalConfig.set(Key.PARTITION, parsedPartitions);
}
} else {
// 非分区表，则不能配置分区
if (null != userConfiguredPartitions
&& !userConfiguredPartitions.isEmpty()) {
&& !userConfiguredPartitions.isEmpty()) {
throw DataXException.asDataXException(OdpsReaderErrorCode.PARTITION_ERROR,
String.format("分区配置错误，源头表:%s 为非分区表, 您不能配置分区. 请您删除该配置项. ", table.getName()));
MESSAGE_SOURCE.message("odpsreader.6", table.getName()));
}
}
this.originalConfig.set(Constant.PARTITION_COLUMNS, partitionColumns);
if (isPartitionedTable) {
LOG.info("{源头表:{} 的所有分区列是:[{}]}", table.getName(),
StringUtils.join(partitionColumns, ","));
LOG.info(MESSAGE_SOURCE.message("odpsreader.7", table.getName(),
StringUtils.join(partitionColumns, ",")));
}
}
private List<String> expandUserConfiguredPartition(
List<String> allPartitions, List<String> userConfiguredPartitions) {
/**
* 将用户配置的分区(可能是直接的分区配置 dt=20170101, 可能是简单正则dt=201701*, 也可能是区间过滤条件 dt>=20170101 and dt<20170130) 和ODPS
* table所有的分区进行匹配,过滤出用户希望同步的分区集合
*
* @param table odps table
* @param allPartitions odps table所有的分区
* @param userConfiguredPartitions 用户配置的分区
* @param tableOriginalPartitionDepth odps table分区级数(一级分区,二级分区,三级分区等)
* @return 返回过滤出的分区
*/
private List<String> expandUserConfiguredPartition(Table table,
List<String> allPartitions,
List<String> userConfiguredPartitions,
int tableOriginalPartitionDepth) {
UserConfiguredPartitionClassification userConfiguredPartitionClassification = OdpsUtil
.classifyUserConfiguredPartitions(userConfiguredPartitions);
if (userConfiguredPartitionClassification.isIncludeHintPartition()) {
List<String> expandUserConfiguredPartitionResult = new ArrayList<String>();
// 处理不包含/*query*/的分区过滤
if (!userConfiguredPartitionClassification.getUserConfiguredNormalPartition().isEmpty()) {
expandUserConfiguredPartitionResult.addAll(expandNoHintUserConfiguredPartition(allPartitions,
userConfiguredPartitionClassification.getUserConfiguredNormalPartition(),
tableOriginalPartitionDepth));
}
if (!allPartitions.isEmpty()) {
expandUserConfiguredPartitionResult.addAll(expandHintUserConfiguredPartition(table,
allPartitions, userConfiguredPartitionClassification.getUserConfiguredHintPartition()));
}
return expandUserConfiguredPartitionResult;
} else {
return expandNoHintUserConfiguredPartition(allPartitions, userConfiguredPartitions,
tableOriginalPartitionDepth);
}
}
/**
* 匹配包含 HINT 条件的过滤
*
* @param table odps table
* @param allPartitions odps table所有的分区
* @param userHintConfiguredPartitions 用户配置的分区
* @return 返回过滤出的分区
*/
private List<String> expandHintUserConfiguredPartition(Table table,
List<String> allPartitions,
List<String> userHintConfiguredPartitions) {
try {
// load odps table all partitions into sqlite memory database
SqliteUtil sqliteUtil = new SqliteUtil();
sqliteUtil.loadAllPartitionsIntoSqlite(table, allPartitions);
return sqliteUtil.selectUserConfiguredPartition(userHintConfiguredPartitions);
} catch (Exception ex) {
throw DataXException.asDataXException(OdpsReaderErrorCode.PARTITION_ERROR,
String.format("Expand user configured partition has exception: %s", ex.getMessage()), ex);
}
}
/**
* 匹配没有 HINT 条件的过滤,包括 简单正则匹配(dt=201701*) 和 直接匹配(dt=20170101)
*
* @param allPartitions odps table所有的分区
* @param userNormalConfiguredPartitions 用户配置的分区
* @param tableOriginalPartitionDepth odps table分区级数(一级分区,二级分区,三级分区等)
* @return 返回过滤出的分区
*/
private List<String> expandNoHintUserConfiguredPartition(List<String> allPartitions,
List<String> userNormalConfiguredPartitions,
int tableOriginalPartitionDepth) {
// 对odps 本身的所有分区进行特殊字符的处理
LOG.info("format partition with rules: remove all space; remove all '; replace / to ,");
// 表里面已有分区量比较大，有些任务无关，没有打印
List<String> allStandardPartitions = OdpsUtil
.formatPartitions(allPartitions);
.formatPartitions(allPartitions);
// 对用户自身配置的所有分区进行特殊字符的处理
List<String> allStandardUserConfiguredPartitions = OdpsUtil
.formatPartitions(userConfiguredPartitions);
.formatPartitions(userNormalConfiguredPartitions);
LOG.info("user configured partition: {}", JSON.toJSONString(userNormalConfiguredPartitions));
LOG.info("formated partition: {}", JSON.toJSONString(allStandardUserConfiguredPartitions));
/**
* 对配置的分区级数(深度)进行检查
@@ -177,20 +261,20 @@ public class OdpsReader extends Reader {
comparedPartitionDepth = comparedPartition.split(",").length;
if (comparedPartitionDepth != firstPartitionDepth) {
throw DataXException.asDataXException(OdpsReaderErrorCode.PARTITION_ERROR,
String.format("分区配置错误, 您所配置的分区级数和该表的实际情况不一致, 比如分区:[%s] 是 %s 级分区, 而分区:[%s] 是 %s 级分区. DataX 是通过英文逗号判断您所配置的分区级数的. 正确的格式形如\"pt=${bizdate}, type=0\" ，请您参考示例修改该配置项. ",
firstPartition, firstPartitionDepth, comparedPartition, comparedPartitionDepth));
MESSAGE_SOURCE
.message("odpsreader.8", firstPartition, firstPartitionDepth, comparedPartition,
comparedPartitionDepth));
}
}
int tableOriginalPartitionDepth = allStandardPartitions.get(0).split(",").length;
if (firstPartitionDepth != tableOriginalPartitionDepth) {
throw DataXException.asDataXException(OdpsReaderErrorCode.PARTITION_ERROR,
String.format("分区配置错误, 您所配置的分区:%s 的级数:%s 与您要读取的 ODPS 源头表的分区级数:%s 不相等. DataX 是通过英文逗号判断您所配置的分区级数的.正确的格式形如\"pt=${bizdate}, type=0\" ，请您参考示例修改该配置项.",
firstPartition, firstPartitionDepth, tableOriginalPartitionDepth));
MESSAGE_SOURCE
.message("odpsreader.9", firstPartition, firstPartitionDepth, tableOriginalPartitionDepth));
}
List<String> retPartitions = FilterUtil.filterByRegulars(allStandardPartitions,
allStandardUserConfiguredPartitions);
allStandardUserConfiguredPartitions);
return retPartitions;
}
@@ -198,11 +282,11 @@ public class OdpsReader extends Reader {
private void dealColumn(Table table) {
// 用户配置的 column 之前已经确保其不为空
List<String> userConfiguredColumns = this.originalConfig.getList(
Key.COLUMN, String.class);
Key.COLUMN, String.class);
List<Column> allColumns = OdpsUtil.getTableAllColumns(table);
List<String> allNormalColumns = OdpsUtil
.getTableOriginalColumnNameList(allColumns);
.getTableOriginalColumnNameList(allColumns);
StringBuilder columnMeta = new StringBuilder();
for (Column column : allColumns) {
@@ -210,26 +294,26 @@ public class OdpsReader extends Reader {
}
columnMeta.setLength(columnMeta.length() - 1);
LOG.info("源头表:{} 的所有字段是:[{}]", table.getName(), columnMeta.toString());
LOG.info(MESSAGE_SOURCE.message("odpsreader.10", table.getName(), columnMeta.toString()));
if (1 == userConfiguredColumns.size()
&& "*".equals(userConfiguredColumns.get(0))) {
LOG.warn("这是一条警告信息，您配置的 ODPS 读取的列为*，这是不推荐的行为，因为当您的表字段个数、类型有变动时，可能影响任务正确性甚至会运行出错. 建议您把所有需要抽取的列都配置上. ");
&& "*".equals(userConfiguredColumns.get(0))) {
LOG.warn(MESSAGE_SOURCE.message("odpsreader.11"));
this.originalConfig.set(Key.COLUMN, allNormalColumns);
}
userConfiguredColumns = this.originalConfig.getList(
Key.COLUMN, String.class);
Key.COLUMN, String.class);
/**
* warn: 字符串常量需要与表原生字段tableOriginalColumnNameList 分开存放 demo:
* ["id","'id'","name"]
*/
List<String> allPartitionColumns = this.originalConfig.getList(
Constant.PARTITION_COLUMNS, String.class);
Constant.PARTITION_COLUMNS, String.class);
List<Pair<String, ColumnType>> parsedColumns = OdpsUtil
.parseColumns(allNormalColumns, allPartitionColumns,
userConfiguredColumns);
.parseColumns(allNormalColumns, allPartitionColumns,
userConfiguredColumns);
this.originalConfig.set(Constant.PARSED_COLUMNS, parsedColumns);
@@ -238,7 +322,7 @@ public class OdpsReader extends Reader {
for (int i = 0, len = parsedColumns.size(); i < len; i++) {
Pair<String, ColumnType> pair = parsedColumns.get(i);
sb.append(String.format(" %s : %s", pair.getLeft(),
pair.getRight()));
pair.getRight()));
if (i != len - 1) {
sb.append(",");
}
@@ -247,9 +331,36 @@ public class OdpsReader extends Reader {
LOG.info("parsed column details: {} .", sb.toString());
}
@Override
public void prepare() {
List<String> preSqls = this.originalConfig.getList(Key.PRE_SQL, String.class);
if (preSqls != null && !preSqls.isEmpty()) {
LOG.info(
String.format("Beigin to exectue preSql : %s. \n Attention: these preSqls must be idempotent!!!",
JSON.toJSONString(preSqls)));
long beginTime = System.currentTimeMillis();
StringBuffer preSqlBuffer = new StringBuffer();
for (String preSql : preSqls) {
preSql = preSql.trim();
if (StringUtils.isNotBlank(preSql) && !preSql.endsWith(";")) {
preSql = String.format("%s;", preSql);
}
if (StringUtils.isNotBlank(preSql)) {
preSqlBuffer.append(preSql);
}
}
if (StringUtils.isNotBlank(preSqlBuffer.toString())) {
OdpsUtil.runSqlTaskWithRetry(this.odps, preSqlBuffer.toString(), "preSql");
} else {
LOG.info("skip to execute the preSql: {}", JSON.toJSONString(preSqls));
}
long endTime = System.currentTimeMillis();
LOG.info(
String.format("Exectue odpsreader preSql successfully! cost time: %s ms.", (endTime - beginTime)));
}
this.initOdpsTableInfo();
}
@Override
@@ -259,6 +370,33 @@ public class OdpsReader extends Reader {
@Override
public void post() {
List<String> postSqls = this.originalConfig.getList(Key.POST_SQL, String.class);
if (postSqls != null && !postSqls.isEmpty()) {
LOG.info(
String.format("Beigin to exectue postSql : %s. \n Attention: these postSqls must be idempotent!!!",
JSON.toJSONString(postSqls)));
long beginTime = System.currentTimeMillis();
StringBuffer postSqlBuffer = new StringBuffer();
for (String postSql : postSqls) {
postSql = postSql.trim();
if (StringUtils.isNotBlank(postSql) && !postSql.endsWith(";")) {
postSql = String.format("%s;", postSql);
}
if (StringUtils.isNotBlank(postSql)) {
postSqlBuffer.append(postSql);
}
}
if (StringUtils.isNotBlank(postSqlBuffer.toString())) {
OdpsUtil.runSqlTaskWithRetry(this.odps, postSqlBuffer.toString(), "postSql");
} else {
LOG.info("skip to execute the postSql: {}", JSON.toJSONString(postSqls));
}
long endTime = System.currentTimeMillis();
LOG.info(
String.format("Exectue odpsreader postSql successfully! cost time: %s ms.", (endTime - beginTime)));
}
}
@Override
@@ -268,6 +406,7 @@ public class OdpsReader extends Reader {
public static class Task extends Reader.Task {
private static final Logger LOG = LoggerFactory.getLogger(Task.class);
private static final MessageSource MESSAGE_SOURCE = MessageSource.loadResourceBundle(OdpsReader.class);
private Configuration readerSliceConf;
private String tunnelServer;
@@ -278,32 +417,35 @@ public class OdpsReader extends Reader {
private boolean isPartitionedTable;
private String sessionId;
private boolean isCompress;
private boolean successOnNoPartition;
@Override
public void init() {
this.readerSliceConf = super.getPluginJobConf();
this.tunnelServer = this.readerSliceConf.getString(
Key.TUNNEL_SERVER, null);
Key.TUNNEL_SERVER, null);
this.odps = OdpsUtil.initOdps(this.readerSliceConf);
this.projectName = this.readerSliceConf.getString(Key.PROJECT);
this.tableName = this.readerSliceConf.getString(Key.TABLE);
this.table = OdpsUtil.getTable(this.odps, projectName, tableName);
this.isPartitionedTable = this.readerSliceConf
.getBool(Constant.IS_PARTITIONED_TABLE);
.getBool(Constant.IS_PARTITIONED_TABLE);
this.sessionId = this.readerSliceConf.getString(Constant.SESSION_ID, null);
this.isCompress = this.readerSliceConf.getBool(Key.IS_COMPRESS, false);
this.successOnNoPartition = this.readerSliceConf.getBool(Key.SUCCESS_ON_NO_PATITION, false);
// sessionId 为空的情况是：切分级别只到 partition 的情况
if (StringUtils.isBlank(this.sessionId)) {
String partition = this.readerSliceConf.getString(Key.PARTITION);
// 没有分区读取时, 是没有sessionId这些的
if (this.isPartitionedTable && StringUtils.isBlank(partition) && this.successOnNoPartition) {
LOG.warn("Partition is blank, but you config successOnNoPartition[true] ,don't need to create session");
} else if (StringUtils.isBlank(this.sessionId)) {
DownloadSession session = OdpsUtil.createMasterSessionForPartitionedTable(odps,
tunnelServer, projectName, tableName, this.readerSliceConf.getString(Key.PARTITION));
tunnelServer, projectName, tableName, this.readerSliceConf.getString(Key.PARTITION));
this.sessionId = session.getId();
}
LOG.info("sessionId:{}", this.sessionId);
}
@@ -316,68 +458,72 @@ public class OdpsReader extends Reader {
DownloadSession downloadSession = null;
String partition = this.readerSliceConf.getString(Key.PARTITION);
if (this.isPartitionedTable && StringUtils.isBlank(partition) && this.successOnNoPartition) {
LOG.warn(String.format(
"Partition is blank,not need to be read"));
recordSender.flush();
return;
}
if (this.isPartitionedTable) {
downloadSession = OdpsUtil.getSlaveSessionForPartitionedTable(this.odps, this.sessionId,
this.tunnelServer, this.projectName, this.tableName, partition);
this.tunnelServer, this.projectName, this.tableName, partition);
} else {
downloadSession = OdpsUtil.getSlaveSessionForNonPartitionedTable(this.odps, this.sessionId,
this.tunnelServer, this.projectName, this.tableName);
this.tunnelServer, this.projectName, this.tableName);
}
long start = this.readerSliceConf.getLong(Constant.START_INDEX, 0);
long count = this.readerSliceConf.getLong(Constant.STEP_COUNT,
downloadSession.getRecordCount());
downloadSession.getRecordCount());
if (count > 0) {
LOG.info(String.format(
"Begin to read ODPS table:%s, partition:%s, startIndex:%s, count:%s.",
this.tableName, partition, start, count));
"Begin to read ODPS table:%s, partition:%s, startIndex:%s, count:%s.",
this.tableName, partition, start, count));
} else if (count == 0) {
LOG.warn(String.format("源头表:%s 的分区:%s 没有内容可抽取, 请您知晓.",
this.tableName, partition));
LOG.warn(MESSAGE_SOURCE.message("odpsreader.12", this.tableName, partition));
return;
} else {
throw DataXException.asDataXException(OdpsReaderErrorCode.READ_DATA_FAIL,
String.format("源头表:%s 的分区:%s 读取行数为负数, 请联系 ODPS 管理员查看表状态!",
this.tableName, partition));
MESSAGE_SOURCE.message("odpsreader.13", this.tableName, partition));
}
TableSchema tableSchema = this.table.getSchema();
Set<Column> allColumns = new HashSet<Column>();
allColumns.addAll(tableSchema.getColumns());
allColumns.addAll(tableSchema.getPartitionColumns());
Map<String, OdpsType> columnTypeMap = new HashMap<String, OdpsType>();
Map<String, TypeInfo> columnTypeMap = new HashMap<String, TypeInfo>();
for (Column column : allColumns) {
columnTypeMap.put(column.getName(), column.getType());
columnTypeMap.put(column.getName(), column.getTypeInfo());
}
try {
List<Configuration> parsedColumnsTmp = this.readerSliceConf
.getListConfiguration(Constant.PARSED_COLUMNS);
.getListConfiguration(Constant.PARSED_COLUMNS);
List<Pair<String, ColumnType>> parsedColumns = new ArrayList<Pair<String, ColumnType>>();
for (int i = 0; i < parsedColumnsTmp.size(); i++) {
Configuration eachColumnConfig = parsedColumnsTmp.get(i);
String columnName = eachColumnConfig.getString("left");
ColumnType columnType = ColumnType
.asColumnType(eachColumnConfig.getString("right"));
.asColumnType(eachColumnConfig.getString("right"));
parsedColumns.add(new MutablePair<String, ColumnType>(
columnName, columnType));
columnName, columnType));
}
ReaderProxy readerProxy = new ReaderProxy(recordSender, downloadSession,
columnTypeMap, parsedColumns, partition, this.isPartitionedTable,
start, count, this.isCompress);
start, count, this.isCompress, this.readerSliceConf);
readerProxy.doRead();
} catch (Exception e) {
throw DataXException.asDataXException(OdpsReaderErrorCode.READ_DATA_FAIL,
String.format("源头表:%s 的分区:%s 读取失败, 请联系 ODPS 管理员查看错误详情.", this.tableName, partition), e);
MESSAGE_SOURCE.message("odpsreader.14", this.tableName, partition), e);
}
}
@Override
public void post() {
}
odpsreader/src/main/java/com/alibaba/datax/plugin/reader/odpsreader/OdpsReaderErrorCode.java
+31 -20
View File
@@ -1,45 +1,53 @@
package com.alibaba.datax.plugin.reader.odpsreader;
import com.alibaba.datax.common.spi.ErrorCode;
import com.alibaba.datax.common.util.MessageSource;
public enum OdpsReaderErrorCode implements ErrorCode {
REQUIRED_VALUE("OdpsReader-00", "您缺失了必须填写的参数值."),
ILLEGAL_VALUE("OdpsReader-01", "您配置的值不合法."),
CREATE_DOWNLOADSESSION_FAIL("OdpsReader-03", "创建 ODPS 的 downloadSession 失败."),
GET_DOWNLOADSESSION_FAIL("OdpsReader-04", "获取 ODPS 的 downloadSession 失败."),
READ_DATA_FAIL("OdpsReader-05", "读取 ODPS 源头表失败."),
GET_ID_KEY_FAIL("OdpsReader-06", "获取 accessId/accessKey 失败."),
REQUIRED_VALUE("DATAX_R_ODPS_001", MessageSource.loadResourceBundle(OdpsReaderErrorCode.class).message("description.DATAX_R_ODPS_001"),MessageSource.loadResourceBundle(OdpsReaderErrorCode.class).message("solution.DATAX_R_ODPS_001")),
ILLEGAL_VALUE("DATAX_R_ODPS_002", MessageSource.loadResourceBundle(OdpsReaderErrorCode.class).message("description.DATAX_R_ODPS_002"),MessageSource.loadResourceBundle(OdpsReaderErrorCode.class).message("solution.DATAX_R_ODPS_002")),
CREATE_DOWNLOADSESSION_FAIL("DATAX_R_ODPS_003", MessageSource.loadResourceBundle(OdpsReaderErrorCode.class).message("description.DATAX_R_ODPS_003"),MessageSource.loadResourceBundle(OdpsReaderErrorCode.class).message("solution.DATAX_R_ODPS_003")),
GET_DOWNLOADSESSION_FAIL("DATAX_R_ODPS_004", MessageSource.loadResourceBundle(OdpsReaderErrorCode.class).message("description.DATAX_R_ODPS_004"),MessageSource.loadResourceBundle(OdpsReaderErrorCode.class).message("solution.DATAX_R_ODPS_004")),
READ_DATA_FAIL("DATAX_R_ODPS_005", MessageSource.loadResourceBundle(OdpsReaderErrorCode.class).message("description.DATAX_R_ODPS_005"),MessageSource.loadResourceBundle(OdpsReaderErrorCode.class).message("solution.DATAX_R_ODPS_005")),
GET_ID_KEY_FAIL("DATAX_R_ODPS_006", MessageSource.loadResourceBundle(OdpsReaderErrorCode.class).message("description.DATAX_R_ODPS_006"),MessageSource.loadResourceBundle(OdpsReaderErrorCode.class).message("solution.DATAX_R_ODPS_006")),
ODPS_READ_EXCEPTION("OdpsReader-07", "读取 odps 异常"),
OPEN_RECORD_READER_FAILED("OdpsReader-08", "打开 recordReader 失败."),
ODPS_READ_EXCEPTION("DATAX_R_ODPS_007", MessageSource.loadResourceBundle(OdpsReaderErrorCode.class).message("description.DATAX_R_ODPS_007"),MessageSource.loadResourceBundle(OdpsReaderErrorCode.class).message("solution.DATAX_R_ODPS_007")),
OPEN_RECORD_READER_FAILED("DATAX_R_ODPS_008", MessageSource.loadResourceBundle(OdpsReaderErrorCode.class).message("description.DATAX_R_ODPS_008"),MessageSource.loadResourceBundle(OdpsReaderErrorCode.class).message("solution.DATAX_R_ODPS_008")),
ODPS_PROJECT_NOT_FOUNT("OdpsReader-10", "您配置的值不合法, odps project 不存在."), //ODPS-0420111: Project not found
ODPS_PROJECT_NOT_FOUNT("DATAX_R_ODPS_009", MessageSource.loadResourceBundle(OdpsReaderErrorCode.class).message("description.DATAX_R_ODPS_009"),MessageSource.loadResourceBundle(OdpsReaderErrorCode.class).message("solution.DATAX_R_ODPS_009")), //ODPS-0420111: Project not found
ODPS_TABLE_NOT_FOUNT("OdpsReader-12", "您配置的值不合法, odps table 不存在."), // ODPS-0130131:Table not found
ODPS_TABLE_NOT_FOUNT("DATAX_R_ODPS_010", MessageSource.loadResourceBundle(OdpsReaderErrorCode.class).message("description.DATAX_R_ODPS_010"),MessageSource.loadResourceBundle(OdpsReaderErrorCode.class).message("solution.DATAX_R_ODPS_010")), // ODPS-0130131:Table not found
ODPS_ACCESS_KEY_ID_NOT_FOUND("OdpsReader-13", "您配置的值不合法, odps accessId,accessKey 不存在."), //ODPS-0410051:Invalid credentials - accessKeyId not found
ODPS_ACCESS_KEY_ID_NOT_FOUND("DATAX_R_ODPS_011", MessageSource.loadResourceBundle(OdpsReaderErrorCode.class).message("description.DATAX_R_ODPS_011"),MessageSource.loadResourceBundle(OdpsReaderErrorCode.class).message("solution.DATAX_R_ODPS_011")), //ODPS-0410051:Invalid credentials - accessKeyId not found
ODPS_ACCESS_KEY_INVALID("OdpsReader-14", "您配置的值不合法, odps accessKey 错误."), //ODPS-0410042:Invalid signature value - User signature dose not match
ODPS_ACCESS_KEY_INVALID("DATAX_R_ODPS_012", MessageSource.loadResourceBundle(OdpsReaderErrorCode.class).message("description.DATAX_R_ODPS_012"),MessageSource.loadResourceBundle(OdpsReaderErrorCode.class).message("solution.DATAX_R_ODPS_012")), //ODPS-0410042:Invalid signature value - User signature dose not match
ODPS_ACCESS_DENY("OdpsReader-15", "拒绝访问, 您不在 您配置的 project 中."), //ODPS-0420095: Access Denied - Authorization Failed [4002], You doesn't exist in project
ODPS_ACCESS_DENY("DATAX_R_ODPS_013", MessageSource.loadResourceBundle(OdpsReaderErrorCode.class).message("description.DATAX_R_ODPS_013"),MessageSource.loadResourceBundle(OdpsReaderErrorCode.class).message("solution.DATAX_R_ODPS_013")), //ODPS-0420095: Access Denied - Authorization Failed [4002], You doesn't exist in project
SPLIT_MODE_ERROR("OdpsReader-30", "splitMode配置错误."),
SPLIT_MODE_ERROR("DATAX_R_ODPS_014", MessageSource.loadResourceBundle(OdpsReaderErrorCode.class).message("description.DATAX_R_ODPS_014"),MessageSource.loadResourceBundle(OdpsReaderErrorCode.class).message("solution.DATAX_R_ODPS_014")),
ACCOUNT_TYPE_ERROR("OdpsReader-31", "odps 账号类型错误."),
ACCOUNT_TYPE_ERROR("DATAX_R_ODPS_015", MessageSource.loadResourceBundle(OdpsReaderErrorCode.class).message("description.DATAX_R_ODPS_015"),MessageSource.loadResourceBundle(OdpsReaderErrorCode.class).message("solution.DATAX_R_ODPS_015")),
VIRTUAL_VIEW_NOT_SUPPORT("OdpsReader-32", "Datax 不支持 读取虚拟视图."),
VIRTUAL_VIEW_NOT_SUPPORT("DATAX_R_ODPS_016", MessageSource.loadResourceBundle(OdpsReaderErrorCode.class).message("description.DATAX_R_ODPS_016"),MessageSource.loadResourceBundle(OdpsReaderErrorCode.class).message("solution.DATAX_R_ODPS_016")),
PARTITION_ERROR("OdpsReader-33", "分区配置错误."),
PARTITION_ERROR("DATAX_R_ODPS_017", MessageSource.loadResourceBundle(OdpsReaderErrorCode.class).message("description.DATAX_R_ODPS_017"),MessageSource.loadResourceBundle(OdpsReaderErrorCode.class).message("solution.DATAX_R_ODPS_017")),
PARTITION_NOT_EXISTS_ERROR("DATAX_R_ODPS_018", MessageSource.loadResourceBundle(OdpsReaderErrorCode.class).message("description.DATAX_R_ODPS_018"),MessageSource.loadResourceBundle(OdpsReaderErrorCode.class).message("solution.DATAX_R_ODPS_018")),
RUN_SQL_FAILED("DATAX_R_ODPS_019", MessageSource.loadResourceBundle(OdpsReaderErrorCode.class).message("description.DATAX_R_ODPS_019"),MessageSource.loadResourceBundle(OdpsReaderErrorCode.class).message("solution.DATAX_R_ODPS_019")),
RUN_SQL_ODPS_EXCEPTION("DATAX_R_ODPS_020", MessageSource.loadResourceBundle(OdpsReaderErrorCode.class).message("description.DATAX_R_ODPS_020"),MessageSource.loadResourceBundle(OdpsReaderErrorCode.class).message("solution.DATAX_R_ODPS_020")),
;
private final String code;
private final String description;
private final String solution;
private OdpsReaderErrorCode(String code, String description) {
private OdpsReaderErrorCode(String code, String description,String solution) {
this.code = code;
this.description = description;
this.solution = solution;
}
@Override
@@ -52,9 +60,12 @@ public enum OdpsReaderErrorCode implements ErrorCode {
return this.description;
}
public String getSolution() {
return solution;
}
@Override
public String toString() {
return String.format("Code:[%s], Description:[%s]. ", this.code,
this.description);
return String.format("Code:%s:%s, Solution:[%s]. ", this.code,this.description,this.solution);
}
}
odpsreader/src/main/java/com/alibaba/datax/plugin/reader/odpsreader/ReaderProxy.java
+726 -85
View File
@@ -3,28 +3,37 @@ package com.alibaba.datax.plugin.reader.odpsreader;
import com.alibaba.datax.common.element.*;
import com.alibaba.datax.common.exception.DataXException;
import com.alibaba.datax.common.plugin.RecordSender;
import com.alibaba.datax.common.util.Configuration;
import com.alibaba.datax.common.util.MessageSource;
import com.alibaba.datax.plugin.reader.odpsreader.util.OdpsUtil;
import com.alibaba.fastjson.JSON;
import com.aliyun.odps.Column;
import com.aliyun.odps.OdpsType;
import com.aliyun.odps.data.*;
import com.aliyun.odps.data.Record;
import com.aliyun.odps.data.RecordReader;
import com.aliyun.odps.tunnel.TableTunnel;
import com.aliyun.odps.type.ArrayTypeInfo;
import com.aliyun.odps.type.MapTypeInfo;
import com.aliyun.odps.type.TypeInfo;
import org.apache.commons.codec.binary.Base64;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.tuple.Pair;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.text.ParseException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.text.SimpleDateFormat;
import java.util.*;
public class ReaderProxy {
private static final Logger LOG = LoggerFactory
.getLogger(ReaderProxy.class);
private static final MessageSource MESSAGE_SOURCE = MessageSource.loadResourceBundle(ReaderProxy.class);
private static boolean IS_DEBUG = LOG.isDebugEnabled();
private RecordSender recordSender;
private TableTunnel.DownloadSession downloadSession;
private Map<String, OdpsType> columnTypeMap;
private Map<String, TypeInfo> columnTypeMap;
private List<Pair<String, ColumnType>> parsedColumns;
private String partition;
private boolean isPartitionTable;
@@ -32,11 +41,38 @@ public class ReaderProxy {
private long start;
private long count;
private boolean isCompress;
private static final String NULL_INDICATOR = null;
// TODO 没有支持用户可配置
// TODO 没有timezone
private SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
// 读取 jvm 默认时区
private Calendar calendarForDate = null;
private boolean useDateWithCalendar = true;
private Calendar initCalendar(Configuration config) {
// 理论上不会有其他选择，有配置化可以随时应急
String calendarType = config.getString("calendarType", "iso8601");
Boolean lenient = config.getBool("calendarLenient", true);
// 默认jvm时区
TimeZone timeZone = TimeZone.getDefault();
String timeZoneStr = config.getString("calendarTimeZone");
if (StringUtils.isNotBlank(timeZoneStr)) {
// 如果用户明确指定使用用户指定的
timeZone = TimeZone.getTimeZone(timeZoneStr);
}
Calendar calendarForDate = new Calendar.Builder().setCalendarType(calendarType).setLenient(lenient)
.setTimeZone(timeZone).build();
return calendarForDate;
}
public ReaderProxy(RecordSender recordSender, TableTunnel.DownloadSession downloadSession,
Map<String, OdpsType> columnTypeMap,
List<Pair<String, ColumnType>> parsedColumns, String partition,
boolean isPartitionTable, long start, long count, boolean isCompress) {
Map<String, TypeInfo> columnTypeMap,
List<Pair<String, ColumnType>> parsedColumns, String partition,
boolean isPartitionTable, long start, long count, boolean isCompress, Configuration taskConfig) {
this.recordSender = recordSender;
this.downloadSession = downloadSession;
this.columnTypeMap = columnTypeMap;
@@ -46,14 +82,24 @@ public class ReaderProxy {
this.start = start;
this.count = count;
this.isCompress = isCompress;
this.calendarForDate = this.initCalendar(taskConfig);
this.useDateWithCalendar = taskConfig.getBool("useDateWithCalendar", true);
}
// warn: odps 分区列和正常列不能重名, 所有列都不不区分大小写
public void doRead() {
try {
LOG.info("start={}, count={}",start, count);
//RecordReader recordReader = downloadSession.openRecordReader(start, count, isCompress);
RecordReader recordReader = OdpsUtil.getRecordReader(downloadSession, start, count, isCompress);
List<Column> userConfigNormalColumns = OdpsUtil.getNormalColumns(this.parsedColumns, this.columnTypeMap);
RecordReader recordReader = null;
// fix #ODPS-52184/10332469, updateColumnsSize表示如果用户指定的读取源表列数100列以内的话,则进行列裁剪优化;
int updateColumnsSize = 100;
if(userConfigNormalColumns.size() <= updateColumnsSize){
recordReader = OdpsUtil.getRecordReader(downloadSession, start, count, isCompress, userConfigNormalColumns);
} else {
recordReader = OdpsUtil.getRecordReader(downloadSession, start, count, isCompress);
}
Record odpsRecord;
Map<String, String> partitionMap = this
@@ -72,7 +118,7 @@ public class ReaderProxy {
} catch (InterruptedException ignored) {
}
recordReader = downloadSession.openRecordReader(start, count, isCompress);
LOG.warn("odps-read-exception, 重试第{}次", retryTimes);
LOG.warn(MESSAGE_SOURCE.message("readerproxy.1", retryTimes));
retryTimes++;
continue;
} else {
@@ -144,9 +190,7 @@ public class ReaderProxy {
throw DataXException
.asDataXException(
OdpsReaderErrorCode.ILLEGAL_VALUE,
String.format(
"您的分区 [%s] 解析出现错误,解析后正确的配置方式类似为 [ pt=1,dt=1 ].",
eachPartition));
MESSAGE_SOURCE.message("readerproxy.2", eachPartition));
}
// warn: translate to lower case, it's more comfortable to
// compare whit user's input columns
@@ -168,8 +212,7 @@ public class ReaderProxy {
partitionColumnName = partitionColumnName.toLowerCase();
// it's will never happen, but add this checking
if (!partitionMap.containsKey(partitionColumnName)) {
String errorMessage = String.format(
"表所有分区信息为: %s 其中找不到 [%s] 对应的分区值.",
String errorMessage = MESSAGE_SOURCE.message("readerproxy.3",
com.alibaba.fastjson.JSON.toJSONString(partitionMap),
partitionColumnName);
throw DataXException.asDataXException(
@@ -190,7 +233,7 @@ public class ReaderProxy {
* every line record of odps table
* @param dataXRecord
* every datax record, to be send to writer. method getXXX() case sensitive
* @param type
* @param typeInfo
* odps column type
* @param columnNameValue
* for partition column it's column value, for normal column it's
@@ -199,83 +242,681 @@ public class ReaderProxy {
* true means partition column and false means normal column
* */
private void odpsColumnToDataXField(Record odpsRecord,
com.alibaba.datax.common.element.Record dataXRecord, OdpsType type,
com.alibaba.datax.common.element.Record dataXRecord, TypeInfo typeInfo,
String columnNameValue, boolean isPartitionColumn) {
ArrayRecord record = (ArrayRecord) odpsRecord;
OdpsType type = typeInfo.getOdpsType();
switch (type) {
case BIGINT: {
if (isPartitionColumn) {
dataXRecord.addColumn(new LongColumn(columnNameValue));
} else {
dataXRecord.addColumn(new LongColumn(odpsRecord
.getBigint(columnNameValue)));
}
break;
}
case BOOLEAN: {
if (isPartitionColumn) {
dataXRecord.addColumn(new BoolColumn(columnNameValue));
} else {
dataXRecord.addColumn(new BoolColumn(odpsRecord
.getBoolean(columnNameValue)));
}
break;
}
case DATETIME: {
if (isPartitionColumn) {
try {
dataXRecord.addColumn(new DateColumn(ColumnCast
.string2Date(new StringColumn(columnNameValue))));
} catch (ParseException e) {
LOG.error(String.format("", this.partition));
String errMessage = String.format(
"您读取分区 [%s] 出现日期转换异常, 日期的字符串表示为 [%s].",
this.partition, columnNameValue);
LOG.error(errMessage);
throw DataXException.asDataXException(
OdpsReaderErrorCode.READ_DATA_FAIL, errMessage, e);
case BIGINT: {
if (isPartitionColumn) {
dataXRecord.addColumn(new LongColumn(columnNameValue));
} else {
dataXRecord.addColumn(new LongColumn(record
.getBigint(columnNameValue)));
}
} else {
dataXRecord.addColumn(new DateColumn(odpsRecord
.getDatetime(columnNameValue)));
break;
}
case BOOLEAN: {
if (isPartitionColumn) {
dataXRecord.addColumn(new BoolColumn(columnNameValue));
} else {
dataXRecord.addColumn(new BoolColumn(record
.getBoolean(columnNameValue)));
}
break;
}
case DATE:
case DATETIME: {
// odps分区列，目前支持TINYINT、SMALLINT、INT、BIGINT、VARCHAR和STRING类型
if (isPartitionColumn) {
try {
dataXRecord.addColumn(new DateColumn(ColumnCast
.string2Date(new StringColumn(columnNameValue))));
} catch (ParseException e) {
String errMessage = MESSAGE_SOURCE.message("readerproxy.4",
this.partition, columnNameValue);
LOG.error(errMessage);
throw DataXException.asDataXException(
OdpsReaderErrorCode.READ_DATA_FAIL, errMessage, e);
}
} else {
if (com.aliyun.odps.OdpsType.DATETIME == type) {
dataXRecord.addColumn(new DateColumn(record
.getDatetime(columnNameValue)));
} else {
if (this.useDateWithCalendar) {
dataXRecord.addColumn(new DateColumn(record.
getDate(columnNameValue, this.calendarForDate)));
} else {
dataXRecord.addColumn(new DateColumn(record
.getDate(columnNameValue)));
}
}
}
break;
}
case DOUBLE: {
if (isPartitionColumn) {
dataXRecord.addColumn(new DoubleColumn(columnNameValue));
} else {
dataXRecord.addColumn(new DoubleColumn(odpsRecord
.getDouble(columnNameValue)));
break;
}
break;
}
case DECIMAL: {
if(isPartitionColumn) {
dataXRecord.addColumn(new DoubleColumn(columnNameValue));
} else {
dataXRecord.addColumn(new DoubleColumn(odpsRecord.getDecimal(columnNameValue)));
case DOUBLE: {
if (isPartitionColumn) {
dataXRecord.addColumn(new DoubleColumn(columnNameValue));
} else {
dataXRecord.addColumn(new DoubleColumn(record
.getDouble(columnNameValue)));
}
break;
}
break;
}
case STRING: {
if (isPartitionColumn) {
dataXRecord.addColumn(new StringColumn(columnNameValue));
} else {
dataXRecord.addColumn(new StringColumn(odpsRecord
.getString(columnNameValue)));
case DECIMAL: {
if(isPartitionColumn) {
dataXRecord.addColumn(new DoubleColumn(columnNameValue));
} else {
dataXRecord.addColumn(new DoubleColumn(record.getDecimal(columnNameValue)));
}
break;
}
break;
}
default:
throw DataXException
.asDataXException(
OdpsReaderErrorCode.ILLEGAL_VALUE,
String.format(
"DataX 抽取 ODPS 数据不支持字段类型为:[%s]. 目前支持抽取的字段类型有：bigint, boolean, datetime, double, decimal, string. "
+ "您可以选择不抽取 DataX 不支持的字段或者联系 ODPS 管理员寻求帮助.",
type));
case STRING: {
if (isPartitionColumn) {
dataXRecord.addColumn(new StringColumn(columnNameValue));
} else {
dataXRecord.addColumn(new StringColumn(record
.getString(columnNameValue)));
}
break;
}
case TINYINT:
if (isPartitionColumn) {
dataXRecord.addColumn(new LongColumn(columnNameValue));
} else {
Byte value = record.getTinyint(columnNameValue);
Integer intValue = value != null ? value.intValue() : null;
dataXRecord.addColumn(new LongColumn(intValue));
}
break;
case SMALLINT: {
if (isPartitionColumn) {
dataXRecord.addColumn(new LongColumn(columnNameValue));
} else {
Short value = record.getSmallint(columnNameValue);
Long valueInLong = null;
if (null != value) {
valueInLong = value.longValue();
}
dataXRecord.addColumn(new LongColumn(valueInLong));
}
break;
}
case INT: {
if (isPartitionColumn) {
dataXRecord.addColumn(new LongColumn(columnNameValue));
} else {
dataXRecord.addColumn(new LongColumn(record
.getInt(columnNameValue)));
}
break;
}
case FLOAT: {
if (isPartitionColumn) {
dataXRecord.addColumn(new DoubleColumn(columnNameValue));
} else {
dataXRecord.addColumn(new DoubleColumn(record
.getFloat(columnNameValue)));
}
break;
}
case VARCHAR: {
if (isPartitionColumn) {
dataXRecord.addColumn(new StringColumn(columnNameValue));
} else {
Varchar value = record.getVarchar(columnNameValue);
String columnValue = value != null ? value.getValue() : null;
dataXRecord.addColumn(new StringColumn(columnValue));
}
break;
}
case TIMESTAMP: {
if (isPartitionColumn) {
try {
dataXRecord.addColumn(new DateColumn(ColumnCast
.string2Date(new StringColumn(columnNameValue))));
} catch (ParseException e) {
String errMessage = MESSAGE_SOURCE.message("readerproxy.4",
this.partition, columnNameValue);
LOG.error(errMessage);
throw DataXException.asDataXException(
OdpsReaderErrorCode.READ_DATA_FAIL, errMessage, e);
}
} else {
dataXRecord.addColumn(new DateColumn(record
.getTimestamp(columnNameValue)));
}
break;
}
case BINARY: {
if (isPartitionColumn) {
dataXRecord.addColumn(new BytesColumn(columnNameValue.getBytes()));
} else {
// dataXRecord.addColumn(new BytesColumn(record
// .getBinary(columnNameValue).data()));
Binary binaryData = record.getBinary(columnNameValue);
if (null == binaryData) {
dataXRecord.addColumn(new BytesColumn(null));
} else {
dataXRecord.addColumn(new BytesColumn(binaryData.data()));
}
}
break;
}
case ARRAY: {
if (isPartitionColumn) {
dataXRecord.addColumn(new StringColumn(columnNameValue));
} else {
List arrayValue = record.getArray(columnNameValue);
if (arrayValue == null) {
dataXRecord.addColumn(new StringColumn(null));
} else {
dataXRecord.addColumn(new StringColumn(JSON.toJSONString(transOdpsArrayToJavaList(arrayValue, (ArrayTypeInfo)typeInfo))));
}
}
break;
}
case MAP: {
if (isPartitionColumn) {
dataXRecord.addColumn(new StringColumn(columnNameValue));
} else {
Map mapValue = record.getMap(columnNameValue);
if (mapValue == null) {
dataXRecord.addColumn(new StringColumn(null));
} else {
dataXRecord.addColumn(new StringColumn(JSON.toJSONString(transOdpsMapToJavaMap(mapValue, (MapTypeInfo)typeInfo))));
}
}
break;
}
case STRUCT: {
if (isPartitionColumn) {
dataXRecord.addColumn(new StringColumn(columnNameValue));
} else {
Struct structValue = record.getStruct(columnNameValue);
if (structValue == null) {
dataXRecord.addColumn(new StringColumn(null));
} else {
dataXRecord.addColumn(new StringColumn(JSON.toJSONString(transOdpsStructToJavaMap(structValue))));
}
}
break;
}
default:
throw DataXException.asDataXException(
OdpsReaderErrorCode.ILLEGAL_VALUE,
MESSAGE_SOURCE.message("readerproxy.5", type));
}
}
private List transOdpsArrayToJavaList(List odpsArray, ArrayTypeInfo typeInfo) {
TypeInfo eleType = typeInfo.getElementTypeInfo();
List result = new ArrayList();
switch (eleType.getOdpsType()) {
// warn:array<double> [1.2, 3.4] 被转为了："["1.2", "3.4"]"， 本来应该被转换成 "[1.2, 3.4]"
// 注意回归Case覆盖
case BIGINT:
case DOUBLE:
case INT:
case FLOAT:
case DECIMAL:
case TINYINT:
case SMALLINT:
for (Object item : odpsArray) {
Object object = item;
result.add(object == null ? NULL_INDICATOR : object);
}
return result;
case BOOLEAN: // 未调整array<Boolean> 问题
case STRING:
case VARCHAR:
case CHAR:
case TIMESTAMP:
case DATE:
for (Object item : odpsArray) {
Object object = item;
result.add(object == null ? NULL_INDICATOR : object.toString());
}
return result;
/**
* 日期类型
*/
case DATETIME:
for (Object item : odpsArray) {
Date dateVal = (Date) item;
result.add(dateVal == null ? NULL_INDICATOR : dateFormat.format(dateVal));
}
return result;
/**
* 字节数组
*/
case BINARY:
for (Object item : odpsArray) {
Binary binaryVal = (Binary) item;
result.add(binaryVal == null ? NULL_INDICATOR :
Base64.encodeBase64(binaryVal.data()));
}
return result;
/**
* 日期间隔
*/
case INTERVAL_DAY_TIME:
for (Object item : odpsArray) {
IntervalDayTime dayTimeVal = (IntervalDayTime) item;
result.add(dayTimeVal == null ? NULL_INDICATOR :
transIntervalDayTimeToJavaMap(dayTimeVal));
}
return result;
/**
* 年份间隔
*/
case INTERVAL_YEAR_MONTH:
for (Object item : odpsArray) {
IntervalYearMonth yearMonthVal = (IntervalYearMonth) item;
result.add(yearMonthVal == null ? NULL_INDICATOR :
transIntervalYearMonthToJavaMap(yearMonthVal));
}
return result;
/**
* 结构体
*/
case STRUCT:
for (Object item : odpsArray) {
Struct structVal = (Struct) item;
result.add(structVal == null ? NULL_INDICATOR :
transOdpsStructToJavaMap(structVal));
}
return result;
/**
* MAP类型
*/
case MAP:
for (Object item : odpsArray) {
Map mapVal = (Map) item;
result.add(mapVal == null ? NULL_INDICATOR :
transOdpsMapToJavaMap(mapVal, (MapTypeInfo) eleType));
}
return result;
/**
* ARRAY类型
*/
case ARRAY:
for (Object item : odpsArray) {
List arrayVal = (List) item;
result.add(arrayVal == null ? NULL_INDICATOR :
transOdpsArrayToJavaList(arrayVal, (ArrayTypeInfo) eleType));
}
return result;
default:
throw new IllegalArgumentException("decode record failed. column type: " + eleType.getTypeName());
}
}
private Map transOdpsMapToJavaMap(Map odpsMap, MapTypeInfo typeInfo) {
TypeInfo keyType = typeInfo.getKeyTypeInfo();
TypeInfo valueType = typeInfo.getValueTypeInfo();
Map result = new HashMap();
Set<Map.Entry> entrySet = null;
switch (valueType.getOdpsType()) {
case BIGINT:
case DOUBLE:
case BOOLEAN:
case STRING:
case DECIMAL:
case TINYINT:
case SMALLINT:
case INT:
case FLOAT:
case CHAR:
case VARCHAR:
case DATE:
case TIMESTAMP:
switch (keyType.getOdpsType()) {
case DATETIME:
entrySet = odpsMap.entrySet();
for (Map.Entry item : entrySet) {
Object value = item.getValue();
result.put(dateFormat.format((Date)item.getKey()), value == null ? NULL_INDICATOR : value.toString());
}
return result;
case BINARY:
entrySet = odpsMap.entrySet();
for (Map.Entry item : entrySet) {
Object value = item.getValue();
result.put(Base64.encodeBase64(((Binary)item.getKey()).data()),
value == null ? NULL_INDICATOR : value.toString());
}
return result;
default:
entrySet = odpsMap.entrySet();
for (Map.Entry item : entrySet) {
Object value = item.getValue();
result.put(item.getKey(), value == null ? NULL_INDICATOR : value.toString());
}
return result;
}
/**
* 日期类型
*/
case DATETIME:
switch (keyType.getOdpsType()) {
case DATETIME:
entrySet = odpsMap.entrySet();
for (Map.Entry item : entrySet) {
Date dateVal = (Date) item.getValue();
result.put(dateFormat.format((Date)item.getKey()),
dateVal == null ? NULL_INDICATOR : dateFormat.format(dateVal));
}
return result;
case BINARY:
entrySet = odpsMap.entrySet();
for (Map.Entry item : entrySet) {
Date dateVal = (Date) item.getValue();
result.put(Base64.encodeBase64(((Binary)item.getKey()).data()),
dateVal == null ? NULL_INDICATOR : dateFormat.format(dateVal));
}
return result;
default:
entrySet = odpsMap.entrySet();
for (Map.Entry item : entrySet) {
Date dateVal = (Date) item.getValue();
result.put(item.getKey(), dateVal == null ? NULL_INDICATOR : dateFormat.format(dateVal));
}
return result;
}
/**
* 字节数组
*/
case BINARY:
switch (keyType.getOdpsType()) {
case DATETIME:
entrySet = odpsMap.entrySet();
for (Map.Entry item : entrySet) {
Binary binaryVal = (Binary) item.getValue();
result.put(dateFormat.format((Date)item.getKey()), binaryVal == null ? NULL_INDICATOR :
Base64.encodeBase64(binaryVal.data()));
}
return result;
case BINARY:
entrySet = odpsMap.entrySet();
for (Map.Entry item : entrySet) {
Binary binaryVal = (Binary) item.getValue();
result.put(Base64.encodeBase64(((Binary)item.getKey()).data()),
binaryVal == null ? NULL_INDICATOR :
Base64.encodeBase64(binaryVal.data()));
}
return result;
default:
entrySet = odpsMap.entrySet();
for (Map.Entry item : entrySet) {
Binary binaryVal = (Binary) item.getValue();
result.put(item.getKey(), binaryVal == null ? NULL_INDICATOR :
Base64.encodeBase64(binaryVal.data()));
}
return result;
}
/**
* 日期间隔
*/
case INTERVAL_DAY_TIME:
switch (keyType.getOdpsType()) {
case DATETIME:
entrySet = odpsMap.entrySet();
for (Map.Entry item : entrySet) {
IntervalDayTime dayTimeVal = (IntervalDayTime) item.getValue();
result.put(dateFormat.format((Date)item.getKey()), dayTimeVal == null ? NULL_INDICATOR :
transIntervalDayTimeToJavaMap(dayTimeVal));
}
return result;
case BINARY:
entrySet = odpsMap.entrySet();
for (Map.Entry item : entrySet) {
IntervalDayTime dayTimeVal = (IntervalDayTime) item.getValue();
result.put(Base64.encodeBase64(((Binary)item.getKey()).data()),
dayTimeVal == null ? NULL_INDICATOR :
transIntervalDayTimeToJavaMap(dayTimeVal));
}
return result;
default:
entrySet = odpsMap.entrySet();
for (Map.Entry item : entrySet) {
IntervalDayTime dayTimeVal = (IntervalDayTime) item.getValue();
result.put(item.getKey(), dayTimeVal == null ? NULL_INDICATOR :
transIntervalDayTimeToJavaMap(dayTimeVal));
}
return result;
}
/**
* 年份间隔
*/
case INTERVAL_YEAR_MONTH:
switch (keyType.getOdpsType()) {
case DATETIME:
entrySet = odpsMap.entrySet();
for (Map.Entry item : entrySet) {
IntervalYearMonth yearMonthVal = (IntervalYearMonth) item.getValue();
result.put(dateFormat.format((Date)item.getKey()), yearMonthVal == null ? NULL_INDICATOR :
transIntervalYearMonthToJavaMap(yearMonthVal));
}
return result;
case BINARY:
entrySet = odpsMap.entrySet();
for (Map.Entry item : entrySet) {
IntervalYearMonth yearMonthVal = (IntervalYearMonth) item.getValue();
result.put(Base64.encodeBase64(((Binary)item.getKey()).data()),
yearMonthVal == null ? NULL_INDICATOR :
transIntervalYearMonthToJavaMap(yearMonthVal));
}
return result;
default:
entrySet = odpsMap.entrySet();
for (Map.Entry item : entrySet) {
IntervalYearMonth yearMonthVal = (IntervalYearMonth) item.getValue();
result.put(item.getKey(), yearMonthVal == null ? NULL_INDICATOR :
transIntervalYearMonthToJavaMap(yearMonthVal));
}
return result;
}
/**
* 结构体
*/
case STRUCT:
switch (keyType.getOdpsType()) {
case DATETIME:
entrySet = odpsMap.entrySet();
for (Map.Entry item : entrySet) {
Struct structVal = (Struct) item.getValue();
result.put(dateFormat.format((Date)item.getKey()), structVal == null ? NULL_INDICATOR :
transOdpsStructToJavaMap(structVal));
}
return result;
case BINARY:
entrySet = odpsMap.entrySet();
for (Map.Entry item : entrySet) {
Struct structVal = (Struct) item.getValue();
result.put(Base64.encodeBase64(((Binary)item.getKey()).data()),
structVal == null ? NULL_INDICATOR :
transOdpsStructToJavaMap(structVal));
}
return result;
default:
entrySet = odpsMap.entrySet();
for (Map.Entry item : entrySet) {
Struct structVal = (Struct) item.getValue();
result.put(item.getKey(), structVal == null ? NULL_INDICATOR :
transOdpsStructToJavaMap(structVal));
}
return result;
}
/**
* MAP类型
*/
case MAP:
switch (keyType.getOdpsType()) {
case DATETIME:
entrySet = odpsMap.entrySet();
for (Map.Entry item : entrySet) {
Map mapVal = (Map) item.getValue();
result.put(dateFormat.format((Date)item.getKey()),mapVal == null ? NULL_INDICATOR :
transOdpsMapToJavaMap(mapVal, (MapTypeInfo) valueType));
}
return result;
case BINARY:
entrySet = odpsMap.entrySet();
for (Map.Entry item : entrySet) {
Map mapVal = (Map) item.getValue();
result.put(Base64.encodeBase64(((Binary)item.getKey()).data()),
mapVal == null ? NULL_INDICATOR : transOdpsMapToJavaMap(mapVal, (MapTypeInfo) valueType));
}
return result;
default:
entrySet = odpsMap.entrySet();
for (Map.Entry item : entrySet) {
Map mapVal = (Map) item.getValue();
result.put(item.getKey(), mapVal == null ? NULL_INDICATOR :
transOdpsMapToJavaMap(mapVal, (MapTypeInfo) valueType));
}
return result;
}
/**
* ARRAY类型
*/
case ARRAY:
switch (keyType.getOdpsType()) {
case DATETIME:
entrySet = odpsMap.entrySet();
for (Map.Entry item : entrySet) {
List arrayVal = (List) item.getValue();
result.put(dateFormat.format((Date)item.getKey()),arrayVal == null ? NULL_INDICATOR :
transOdpsArrayToJavaList(arrayVal, (ArrayTypeInfo) valueType));
}
return result;
case BINARY:
entrySet = odpsMap.entrySet();
for (Map.Entry item : entrySet) {
List arrayVal = (List) item.getValue();
result.put(Base64.encodeBase64(((Binary)item.getKey()).data()),
arrayVal == null ? NULL_INDICATOR : transOdpsArrayToJavaList(arrayVal, (ArrayTypeInfo) valueType));
}
return result;
default:
entrySet = odpsMap.entrySet();
for (Map.Entry item : entrySet) {
List arrayVal = (List) item.getValue();
result.put(item.getKey(), arrayVal == null ? NULL_INDICATOR :
transOdpsArrayToJavaList(arrayVal, (ArrayTypeInfo) valueType));
}
return result;
}
default:
throw new IllegalArgumentException("decode record failed. column type: " + valueType.getTypeName());
}
}
private Map transIntervalDayTimeToJavaMap(IntervalDayTime dayTime) {
Map<String, Long> result = new HashMap<String, Long>();
result.put("totalSeconds", dayTime.getTotalSeconds());
result.put("nanos", (long)dayTime.getNanos());
return result;
}
private Map transOdpsStructToJavaMap(Struct odpsStruct) {
Map result = new HashMap();
for (int i = 0; i < odpsStruct.getFieldCount(); i++) {
String fieldName = odpsStruct.getFieldName(i);
Object fieldValue = odpsStruct.getFieldValue(i);
TypeInfo fieldType = odpsStruct.getFieldTypeInfo(i);
switch (fieldType.getOdpsType()) {
case BIGINT:
case DOUBLE:
case BOOLEAN:
case STRING:
case DECIMAL:
case TINYINT:
case SMALLINT:
case INT:
case FLOAT:
case VARCHAR:
case CHAR:
case TIMESTAMP:
case DATE:
result.put(fieldName, fieldValue == null ? NULL_INDICATOR : fieldValue.toString());
break;
/**
* 日期类型
*/
case DATETIME:
Date dateVal = (Date) fieldValue;
result.put(fieldName, dateVal == null ? NULL_INDICATOR : dateFormat.format(dateVal));
break;
/**
* 字节数组
*/
case BINARY:
Binary binaryVal = (Binary) fieldValue;
result.put(fieldName, binaryVal == null ? NULL_INDICATOR :
Base64.encodeBase64(binaryVal.data()));
break;
/**
* 日期间隔
*/
case INTERVAL_DAY_TIME:
IntervalDayTime dayTimeVal = (IntervalDayTime) fieldValue;
result.put(fieldName, dayTimeVal == null ? NULL_INDICATOR :
transIntervalDayTimeToJavaMap(dayTimeVal));
break;
/**
* 年份间隔
*/
case INTERVAL_YEAR_MONTH:
IntervalYearMonth yearMonthVal = (IntervalYearMonth) fieldValue;
result.put(fieldName, yearMonthVal == null ? NULL_INDICATOR :
transIntervalYearMonthToJavaMap(yearMonthVal));
break;
/**
* 结构体
*/
case STRUCT:
Struct structVal = (Struct) fieldValue;
result.put(fieldName, structVal == null ? NULL_INDICATOR :
transOdpsStructToJavaMap(structVal));
break;
/**
* MAP类型
*/
case MAP:
Map mapVal = (Map) fieldValue;
result.put(fieldName, mapVal == null ? NULL_INDICATOR :
transOdpsMapToJavaMap(mapVal, (MapTypeInfo) fieldType));
break;
/**
* ARRAY类型
*/
case ARRAY:
List arrayVal = (List) fieldValue;
result.put(fieldName, arrayVal == null ? NULL_INDICATOR :
transOdpsArrayToJavaList(arrayVal, (ArrayTypeInfo) fieldType));
break;
default:
throw new IllegalArgumentException("decode record failed. column type: " + fieldType.getTypeName());
}
}
return result;
}
private Map transIntervalYearMonthToJavaMap(IntervalYearMonth yearMonth) {
Map <String, Integer> result = new HashMap<String, Integer>();
result.put("years", yearMonth.getYears());
result.put("months", yearMonth.getMonths());
return result;
}
}
odpsreader/src/main/java/com/alibaba/datax/plugin/reader/odpsreader/util/DESCipher.java
-355
View File
@@ -1,355 +0,0 @@
/**
* (C) 2010-2014 Alibaba Group Holding Limited.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package com.alibaba.datax.plugin.reader.odpsreader.util;
import javax.crypto.Cipher;
import javax.crypto.SecretKey;
import javax.crypto.SecretKeyFactory;
import javax.crypto.spec.DESKeySpec;
import java.security.SecureRandom;
/**
* 　　* DES加解密,支持与delphi交互(字符串编码需统一为UTF-8)
*
* 　　*
*
* 　　* @author wym
*
* 　　
*/
public class DESCipher {
/**
* 　　* 密钥
*
* 　　
*/
public static final String KEY = "DESDES";
private final static String DES = "DES";
/**
* 　　* 加密
*
* 　　*
*
* 　　* @param src
*
* 　　* 明文(字节)
*
* 　　* @param key
*
* 　　* 密钥，长度必须是8的倍数
*
* 　　* @return 密文(字节)
*
* 　　* @throws Exception
*
* 　　
*/
public static byte[] encrypt(byte[] src, byte[] key) throws Exception {
// DES算法要求有一个可信任的随机数源
SecureRandom sr = new SecureRandom();
// 从原始密匙数据创建DESKeySpec对象
DESKeySpec dks = new DESKeySpec(key);
// 创建一个密匙工厂，然后用它把DESKeySpec转换成
// 一个SecretKey对象
SecretKeyFactory keyFactory = SecretKeyFactory.getInstance(DES);
SecretKey securekey = keyFactory.generateSecret(dks);
// Cipher对象实际完成加密操作
Cipher cipher = Cipher.getInstance(DES);
// 用密匙初始化Cipher对象
cipher.init(Cipher.ENCRYPT_MODE, securekey, sr);
// 现在，获取数据并加密
// 正式执行加密操作
return cipher.doFinal(src);
}
/**
* 　　* 解密
*
* 　　*
*
* 　　* @param src
*
* 　　* 密文(字节)
*
* 　　* @param key
*
* 　　* 密钥，长度必须是8的倍数
*
* 　　* @return 明文(字节)
*
* 　　* @throws Exception
*
* 　　
*/
public static byte[] decrypt(byte[] src, byte[] key) throws Exception {
// DES算法要求有一个可信任的随机数源
SecureRandom sr = new SecureRandom();
// 从原始密匙数据创建一个DESKeySpec对象
DESKeySpec dks = new DESKeySpec(key);
// 创建一个密匙工厂，然后用它把DESKeySpec对象转换成
// 一个SecretKey对象
SecretKeyFactory keyFactory = SecretKeyFactory.getInstance(DES);
SecretKey securekey = keyFactory.generateSecret(dks);
// Cipher对象实际完成解密操作
Cipher cipher = Cipher.getInstance(DES);
// 用密匙初始化Cipher对象
cipher.init(Cipher.DECRYPT_MODE, securekey, sr);
// 现在，获取数据并解密
// 正式执行解密操作
return cipher.doFinal(src);
}
/**
* 　　* 加密
*
* 　　*
*
* 　　* @param src
*
* 　　* 明文(字节)
*
* 　　* @return 密文(字节)
*
* 　　* @throws Exception
*
* 　　
*/
public static byte[] encrypt(byte[] src) throws Exception {
return encrypt(src, KEY.getBytes());
}
/**
* 　　* 解密
*
* 　　*
*
* 　　* @param src
*
* 　　* 密文(字节)
*
* 　　* @return 明文(字节)
*
* 　　* @throws Exception
*
* 　　
*/
public static byte[] decrypt(byte[] src) throws Exception {
return decrypt(src, KEY.getBytes());
}
/**
* 　　* 加密
*
* 　　*
*
* 　　* @param src
*
* 　　* 明文(字符串)
*
* 　　* @return 密文(16进制字符串)
*
* 　　* @throws Exception
*
* 　　
*/
public final static String encrypt(String src) {
try {
return byte2hex(encrypt(src.getBytes(), KEY.getBytes()));
} catch (Exception e) {
e.printStackTrace();
}
return null;
}
/**
* 　　* 解密
*
* 　　*
*
* 　　* @param src
*
* 　　* 密文(字符串)
*
* 　　* @return 明文(字符串)
*
* 　　* @throws Exception
*
* 　　
*/
public final static String decrypt(String src) {
try {
return new String(decrypt(hex2byte(src.getBytes()), KEY.getBytes()));
} catch (Exception e) {
e.printStackTrace();
}
return null;
}
/**
* 　　* 加密
*
* 　　*
*
* 　　* @param src
*
* 　　* 明文(字节)
*
* 　　* @return 密文(16进制字符串)
*
* 　　* @throws Exception
*
* 　　
*/
public static String encryptToString(byte[] src) throws Exception {
return encrypt(new String(src));
}
/**
* 　　* 解密
*
* 　　*
*
* 　　* @param src
*
* 　　* 密文(字节)
*
* 　　* @return 明文(字符串)
*
* 　　* @throws Exception
*
* 　　
*/
public static String decryptToString(byte[] src) throws Exception {
return decrypt(new String(src));
}
public static String byte2hex(byte[] b) {
String hs = "";
String stmp = "";
for (int n = 0; n < b.length; n++) {
stmp = (Integer.toHexString(b[n] & 0XFF));
if (stmp.length() == 1)
hs = hs + "0" + stmp;
else
hs = hs + stmp;
}
return hs.toUpperCase();
}
public static byte[] hex2byte(byte[] b) {
if ((b.length % 2) != 0)
throw new IllegalArgumentException("长度不是偶数");
byte[] b2 = new byte[b.length / 2];
for (int n = 0; n < b.length; n += 2) {
String item = new String(b, n, 2);
b2[n / 2] = (byte) Integer.parseInt(item, 16);
}
return b2;
}
/*
* public static void main(String[] args) { try { String src = "cheetah";
* String crypto = DESCipher.encrypt(src); System.out.println("密文[" + src +
* "]:" + crypto); System.out.println("解密后:" + DESCipher.decrypt(crypto)); }
* catch (Exception e) { e.printStackTrace(); } }
*/
}
odpsreader/src/main/java/com/alibaba/datax/plugin/reader/odpsreader/util/IdAndKeyUtil.java
+9 -29
View File
@@ -1,5 +1,5 @@
/**
* (C) 2010-2014 Alibaba Group Holding Limited.
* (C) 2010-2022 Alibaba Group Holding Limited.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -18,9 +18,11 @@ package com.alibaba.datax.plugin.reader.odpsreader.util;
import com.alibaba.datax.common.exception.DataXException;
import com.alibaba.datax.common.util.Configuration;
import com.alibaba.datax.plugin.reader.odpsreader.Constant;
import com.alibaba.datax.common.util.IdAndKeyRollingUtil;
import com.alibaba.datax.common.util.MessageSource;
import com.alibaba.datax.plugin.reader.odpsreader.Key;
import com.alibaba.datax.plugin.reader.odpsreader.OdpsReaderErrorCode;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -29,6 +31,7 @@ import java.util.Map;
public class IdAndKeyUtil {
private static Logger LOG = LoggerFactory.getLogger(IdAndKeyUtil.class);
private static MessageSource MESSAGE_SOURCE = MessageSource.loadResourceBundle(IdAndKeyUtil.class);
public static Configuration parseAccessIdAndKey(Configuration originalConfig) {
String accessId = originalConfig.getString(Key.ACCESS_ID);
@@ -50,36 +53,13 @@ public class IdAndKeyUtil {
private static Configuration getAccessIdAndKeyFromEnv(Configuration originalConfig,
Map<String, String> envProp) {
String accessId = null;
String accessKey = null;
String skynetAccessID = envProp.get(Constant.SKYNET_ACCESSID);
String skynetAccessKey = envProp.get(Constant.SKYNET_ACCESSKEY);
if (StringUtils.isNotBlank(skynetAccessID)
|| StringUtils.isNotBlank(skynetAccessKey)) {
/**
* 环境变量中，如果存在SKYNET_ACCESSID/SKYNET_ACCESSKEy（只要有其中一个变量，则认为一定是两个都存在的！），
* 则使用其值作为odps的accessId/accessKey(会解密)
*/
LOG.info("Try to get accessId/accessKey from environment.");
accessId = skynetAccessID;
accessKey = DESCipher.decrypt(skynetAccessKey);
if (StringUtils.isNotBlank(accessKey)) {
originalConfig.set(Key.ACCESS_ID, accessId);
originalConfig.set(Key.ACCESS_KEY, accessKey);
LOG.info("Get accessId/accessKey from environment variables successfully.");
} else {
throw DataXException.asDataXException(OdpsReaderErrorCode.GET_ID_KEY_FAIL,
String.format("从环境变量中获取accessId/accessKey 失败, accessId=[%s]", accessId));
}
} else {
// 如果获取到ak，在getAccessIdAndKeyFromEnv中已经设置到originalConfig了
String accessKey = IdAndKeyRollingUtil.getAccessIdAndKeyFromEnv(originalConfig);
if (StringUtils.isBlank(accessKey)) {
// 无处获取（既没有配置在作业中，也没用在环境变量中）
throw DataXException.asDataXException(OdpsReaderErrorCode.GET_ID_KEY_FAIL,
"无法获取到accessId/accessKey. 它们既不存在于您的配置中，也不存在于环境变量中.");
MESSAGE_SOURCE.message("idandkeyutil.2"));
}
return originalConfig;
}
}
odpsreader/src/main/java/com/alibaba/datax/plugin/reader/odpsreader/util/LocalStrings.properties
+25
View File
@@ -0,0 +1,25 @@
descipher.1=\u957F\u5EA6\u4E0D\u662F\u5076\u6570
idandkeyutil.1=\u4ECE\u73AF\u5883\u53D8\u91CF\u4E2D\u83B7\u53D6accessId/accessKey \u5931\u8D25, accessId=[{0}]
idandkeyutil.2=\u65E0\u6CD5\u83B7\u53D6\u5230accessId/accessKey. \u5B83\u4EEC\u65E2\u4E0D\u5B58\u5728\u4E8E\u60A8\u7684\u914D\u7F6E\u4E2D\uFF0C\u4E5F\u4E0D\u5B58\u5728\u4E8E\u73AF\u5883\u53D8\u91CF\u4E2D.
odpssplitutil.1=\u60A8\u6240\u914D\u7F6E\u7684\u5206\u533A\u4E0D\u80FD\u4E3A\u7A7A\u767D.
odpssplitutil.2=\u5207\u5206\u7684 recordCount \u4E0D\u80FD\u4E3A\u8D1F\u6570.recordCount={0}
odpssplitutil.3=\u5207\u5206\u7684 adviceNum \u4E0D\u80FD\u4E3A\u8D1F\u6570.adviceNum={0}
odpssplitutil.4=\u6CE8\u610F: \u7531\u4E8E\u60A8\u914D\u7F6E\u4E86successOnNoPartition\u503C\u4E3Atrue (\u5373\u5F53\u5206\u533A\u503C\u4E0D\u5B58\u5728\u65F6, \u540C\u6B65\u4EFB\u52A1\u4E0D\u62A5\u9519), \u60A8\u8BBE\u7F6E\u7684\u5206\u533A\u65E0\u6CD5\u5339\u914D\u5230ODPS\u8868\u4E2D\u5BF9\u5E94\u7684\u5206\u533A, \u540C\u6B65\u4EFB\u52A1\u7EE7\u7EED...
odpsutil.1=datax\u83B7\u53D6\u4E0D\u5230\u6E90\u8868\u7684\u5217\u4FE1\u606F\uFF0C \u7531\u4E8E\u60A8\u672A\u914D\u7F6E\u8BFB\u53D6\u6E90\u5934\u8868\u7684\u5217\u4FE1\u606F. datax\u65E0\u6CD5\u77E5\u9053\u8BE5\u62BD\u53D6\u8868\u7684\u54EA\u4E9B\u5B57\u6BB5\u7684\u6570\u636E\uFF0C \u6B63\u786E\u7684\u914D\u7F6E\u65B9\u5F0F\u662F\u7ED9 column \u914D\u7F6E\u4E0A\u60A8\u9700\u8981\u8BFB\u53D6\u7684\u5217\u540D\u79F0,\u7528\u82F1\u6587\u9017\u53F7\u5206\u9694.
odpsutil.2=\u60A8\u6240\u914D\u7F6E\u7684maxRetryTime \u503C\u9519\u8BEF. \u8BE5\u503C\u4E0D\u80FD\u5C0F\u4E8E1, \u4E14\u4E0D\u80FD\u5927\u4E8E {0}. \u63A8\u8350\u7684\u914D\u7F6E\u65B9\u5F0F\u662F\u7ED9maxRetryTime \u914D\u7F6E1-11\u4E4B\u95F4\u7684\u67D0\u4E2A\u503C. \u8BF7\u60A8\u68C0\u67E5\u914D\u7F6E\u5E76\u505A\u51FA\u76F8\u5E94\u4FEE\u6539.
odpsutil.3=\u4E0D\u652F\u6301\u7684\u8D26\u53F7\u7C7B\u578B:[{0}]. \u8D26\u53F7\u7C7B\u578B\u76EE\u524D\u4EC5\u652F\u6301aliyun, taobao.
odpsutil.4=\u60A8\u6240\u914D\u7F6E\u7684\u5206\u533A\u4E0D\u80FD\u4E3A\u7A7A\u767D.
odpsutil.5=\u6E90\u5934\u8868\u7684\u5217\u914D\u7F6E\u9519\u8BEF. \u60A8\u6240\u914D\u7F6E\u7684\u5217 [{0}] \u4E0D\u5B58\u5728.
odpsutil.6=open RecordReader\u5931\u8D25. \u8BF7\u8054\u7CFB ODPS \u7BA1\u7406\u5458\u5904\u7406.
odpsutil.7=\u52A0\u8F7D ODPS \u6E90\u5934\u8868:{0} \u5931\u8D25. \u8BF7\u68C0\u67E5\u60A8\u914D\u7F6E\u7684 ODPS \u6E90\u5934\u8868\u7684 [project] \u662F\u5426\u6B63\u786E.
odpsutil.8=\u52A0\u8F7D ODPS \u6E90\u5934\u8868:{0} \u5931\u8D25. \u8BF7\u68C0\u67E5\u60A8\u914D\u7F6E\u7684 ODPS \u6E90\u5934\u8868\u7684 [table] \u662F\u5426\u6B63\u786E.
odpsutil.9=\u52A0\u8F7D ODPS \u6E90\u5934\u8868:{0} \u5931\u8D25. \u8BF7\u68C0\u67E5\u60A8\u914D\u7F6E\u7684 ODPS \u6E90\u5934\u8868\u7684 [accessId] [accessKey]\u662F\u5426\u6B63\u786E.
odpsutil.10=\u52A0\u8F7D ODPS \u6E90\u5934\u8868:{0} \u5931\u8D25. \u8BF7\u68C0\u67E5\u60A8\u914D\u7F6E\u7684 ODPS \u6E90\u5934\u8868\u7684 [accessKey] \u662F\u5426\u6B63\u786E.
odpsutil.11=\u52A0\u8F7D ODPS \u6E90\u5934\u8868:{0} \u5931\u8D25. \u8BF7\u68C0\u67E5\u60A8\u914D\u7F6E\u7684 ODPS \u6E90\u5934\u8868\u7684 [accessId] [accessKey] [project]\u662F\u5426\u5339\u914D.
odpsutil.12=\u52A0\u8F7D ODPS \u6E90\u5934\u8868:{0} \u5931\u8D25. \u8BF7\u68C0\u67E5\u60A8\u914D\u7F6E\u7684 ODPS \u6E90\u5934\u8868\u7684 project,table,accessId,accessKey,odpsServer\u7B49\u503C.
odpsutil.13=\u6267\u884C ODPS SQL\u5931\u8D25, \u8FD4\u56DE\u503C\u4E3A:{0}. \u8BF7\u4ED4\u7EC6\u68C0\u67E5ODPS SQL\u662F\u5426\u6B63\u786E, \u5982\u679C\u68C0\u67E5\u65E0\u8BEF, \u8BF7\u8054\u7CFB ODPS \u503C\u73ED\u540C\u5B66\u5904\u7406. SQL \u5185\u5BB9\u4E3A:[\n{1}\n].
odpsutil.14=\u6267\u884C ODPS SQL \u65F6\u629B\u51FA\u5F02\u5E38, \u8BF7\u4ED4\u7EC6\u68C0\u67E5ODPS SQL\u662F\u5426\u6B63\u786E, \u5982\u679C\u68C0\u67E5\u65E0\u8BEF, \u8BF7\u8054\u7CFB ODPS \u503C\u73ED\u540C\u5B66\u5904\u7406. SQL \u5185\u5BB9\u4E3A:[\n{0}\n].
odpsreader/src/main/java/com/alibaba/datax/plugin/reader/odpsreader/util/OdpsSplitUtil.java
+18 -3
View File
@@ -2,19 +2,26 @@ package com.alibaba.datax.plugin.reader.odpsreader.util;
import com.alibaba.datax.common.exception.DataXException;
import com.alibaba.datax.common.util.Configuration;
import com.alibaba.datax.common.util.MessageSource;
import com.alibaba.datax.common.util.RangeSplitUtil;
import com.alibaba.datax.plugin.reader.odpsreader.Constant;
import com.alibaba.datax.plugin.reader.odpsreader.Key;
import com.alibaba.datax.plugin.reader.odpsreader.OdpsReaderErrorCode;
import com.aliyun.odps.Odps;
import com.aliyun.odps.tunnel.TableTunnel.DownloadSession;
import org.apache.commons.lang3.tuple.ImmutablePair;
import org.apache.commons.lang3.tuple.Pair;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.ArrayList;
import java.util.List;
public final class OdpsSplitUtil {
private static final Logger LOG = LoggerFactory.getLogger(OdpsSplitUtil.class);
private static final MessageSource MESSAGE_SOURCE = MessageSource.loadResourceBundle(OdpsSplitUtil.class);
public static List<Configuration> doSplit(Configuration originalConfig, Odps odps,
int adviceNum) {
@@ -36,9 +43,17 @@ public final class OdpsSplitUtil {
List<String> partitions = originalConfig.getList(Key.PARTITION,
String.class);
if ((null == partitions || partitions.isEmpty()) && originalConfig.getBool(Key.SUCCESS_ON_NO_PATITION, false)) {
Configuration tempConfig = originalConfig.clone();
tempConfig.set(Key.PARTITION, null);
splittedConfigs.add(tempConfig);
LOG.warn(MESSAGE_SOURCE.message("odpssplitutil.4"));
return splittedConfigs;
}
if (null == partitions || partitions.isEmpty()) {
throw DataXException.asDataXException(OdpsReaderErrorCode.ILLEGAL_VALUE,
"您所配置的分区不能为空白.");
MESSAGE_SOURCE.message("odpssplitutil.1"));
}
//splitMode 默认为 record
@@ -141,11 +156,11 @@ public final class OdpsSplitUtil {
*/
private static List<Pair<Long, Long>> splitRecordCount(long recordCount, int adviceNum) {
if(recordCount<0){
throw new IllegalArgumentException("切分的 recordCount 不能为负数.recordCount=" + recordCount);
throw new IllegalArgumentException(MESSAGE_SOURCE.message("odpssplitutil.2", recordCount));
}
if(adviceNum<1){
throw new IllegalArgumentException("切分的 adviceNum 不能为负数.adviceNum=" + adviceNum);
throw new IllegalArgumentException(MESSAGE_SOURCE.message("odpssplitutil.3", adviceNum));
}
List<Pair<Long, Long>> result = new ArrayList<Pair<Long, Long>>();
odpsreader/src/main/java/com/alibaba/datax/plugin/reader/odpsreader/util/OdpsUtil.java
+178 -44
View File
@@ -2,16 +2,22 @@ package com.alibaba.datax.plugin.reader.odpsreader.util;
import com.alibaba.datax.common.exception.DataXException;
import com.alibaba.datax.common.util.Configuration;
import com.alibaba.datax.common.util.DataXCaseEnvUtil;
import com.alibaba.datax.common.util.MessageSource;
import com.alibaba.datax.common.util.RetryUtil;
import com.alibaba.datax.plugin.reader.odpsreader.ColumnType;
import com.alibaba.datax.plugin.reader.odpsreader.Constant;
import com.alibaba.datax.plugin.reader.odpsreader.Key;
import com.alibaba.datax.plugin.reader.odpsreader.OdpsReaderErrorCode;
import com.aliyun.odps.*;
import com.aliyun.odps.Column;
import com.aliyun.odps.account.Account;
import com.aliyun.odps.account.AliyunAccount;
import com.aliyun.odps.account.StsAccount;
import com.aliyun.odps.data.RecordReader;
import com.aliyun.odps.task.SQLTask;
import com.aliyun.odps.tunnel.TableTunnel;
import com.aliyun.odps.type.TypeInfo;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.tuple.MutablePair;
@@ -19,13 +25,12 @@ import org.apache.commons.lang3.tuple.Pair;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.*;
import java.util.concurrent.Callable;
public final class OdpsUtil {
private static final Logger LOG = LoggerFactory.getLogger(OdpsUtil.class);
private static final MessageSource MESSAGE_SOURCE = MessageSource.loadResourceBundle(OdpsUtil.class);
public static int MAX_RETRY_TIME = 10;
@@ -37,8 +42,8 @@ public final class OdpsUtil {
if (null == originalConfig.getList(Key.COLUMN) ||
originalConfig.getList(Key.COLUMN, String.class).isEmpty()) {
throw DataXException.asDataXException(OdpsReaderErrorCode.REQUIRED_VALUE, "datax获取不到源表的列信息， 由于您未配置读取源头表的列信息. datax无法知道该抽取表的哪些字段的数据 " +
"正确的配置方式是给 column 配置上您需要读取的列名称,用英文逗号分隔.");
throw DataXException.asDataXException(OdpsReaderErrorCode.REQUIRED_VALUE,
MESSAGE_SOURCE.message("odpsutil.1"));
}
}
@@ -47,8 +52,8 @@ public final class OdpsUtil {
int maxRetryTime = originalConfig.getInt(Key.MAX_RETRY_TIME,
OdpsUtil.MAX_RETRY_TIME);
if (maxRetryTime < 1 || maxRetryTime > OdpsUtil.MAX_RETRY_TIME) {
throw DataXException.asDataXException(OdpsReaderErrorCode.ILLEGAL_VALUE, "您所配置的maxRetryTime 值错误. 该值不能小于1, 且不能大于 " + OdpsUtil.MAX_RETRY_TIME +
". 推荐的配置方式是给maxRetryTime 配置1-11之间的某个值. 请您检查配置并做出相应修改.");
throw DataXException.asDataXException(OdpsReaderErrorCode.ILLEGAL_VALUE,
MESSAGE_SOURCE.message("odpsutil.2", OdpsUtil.MAX_RETRY_TIME));
}
MAX_RETRY_TIME = maxRetryTime;
}
@@ -59,11 +64,12 @@ public final class OdpsUtil {
String accessId = originalConfig.getString(Key.ACCESS_ID);
String accessKey = originalConfig.getString(Key.ACCESS_KEY);
String project = originalConfig.getString(Key.PROJECT);
String securityToken = originalConfig.getString(Key.SECURITY_TOKEN);
String packageAuthorizedProject = originalConfig.getString(Key.PACKAGE_AUTHORIZED_PROJECT);
String defaultProject;
if(StringUtils.isBlank(packageAuthorizedProject)) {
if (StringUtils.isBlank(packageAuthorizedProject)) {
defaultProject = project;
} else {
defaultProject = packageAuthorizedProject;
@@ -74,21 +80,26 @@ public final class OdpsUtil {
Account account = null;
if (accountType.equalsIgnoreCase(Constant.DEFAULT_ACCOUNT_TYPE)) {
account = new AliyunAccount(accessId, accessKey);
if (StringUtils.isNotBlank(securityToken)) {
account = new StsAccount(accessId, accessKey, securityToken);
} else {
account = new AliyunAccount(accessId, accessKey);
}
} else {
throw DataXException.asDataXException(OdpsReaderErrorCode.ACCOUNT_TYPE_ERROR,
String.format("不支持的账号类型:[%s]. 账号类型目前仅支持aliyun, taobao.", accountType));
MESSAGE_SOURCE.message("odpsutil.3", accountType));
}
Odps odps = new Odps(account);
boolean isPreCheck = originalConfig.getBool("dryRun", false);
if(isPreCheck) {
if (isPreCheck) {
odps.getRestClient().setConnectTimeout(3);
odps.getRestClient().setReadTimeout(3);
odps.getRestClient().setRetryTimes(2);
}
odps.setDefaultProject(defaultProject);
odps.setEndpoint(odpsServer);
odps.setUserAgent("DATAX");
return odps;
}
@@ -103,7 +114,7 @@ public final class OdpsUtil {
table.reload();
return table;
}
}, 3, 1000, false);
}, DataXCaseEnvUtil.getRetryTimes(3), DataXCaseEnvUtil.getRetryInterval(1000), DataXCaseEnvUtil.getRetryExponential(false));
} catch (Exception e) {
throwDataXExceptionWhenReloadTable(e, tableName);
}
@@ -154,7 +165,7 @@ public final class OdpsUtil {
public static String formatPartition(String partition) {
if (StringUtils.isBlank(partition)) {
throw DataXException.asDataXException(OdpsReaderErrorCode.ILLEGAL_VALUE,
"您所配置的分区不能为空白.");
MESSAGE_SOURCE.message("odpsutil.4"));
} else {
return partition.trim().replaceAll(" *= *", "=")
.replaceAll(" */ *", ",").replaceAll(" *, *", ",")
@@ -175,6 +186,35 @@ public final class OdpsUtil {
}
}
/**
* 将用户配置的分区分类成两类:
* (1) 包含 HINT 的区间过滤;
* (2) 不包含 HINT 的普通模式
* @param userConfiguredPartitions
* @return
*/
public static UserConfiguredPartitionClassification classifyUserConfiguredPartitions(List<String> userConfiguredPartitions){
UserConfiguredPartitionClassification userConfiguredPartitionClassification = new UserConfiguredPartitionClassification();
List<String> userConfiguredHintPartition = new ArrayList<String>();
List<String> userConfiguredNormalPartition = new ArrayList<String>();
boolean isIncludeHintPartition = false;
for (String userConfiguredPartition : userConfiguredPartitions){
if (StringUtils.isNotBlank(userConfiguredPartition)){
if (userConfiguredPartition.trim().toLowerCase().startsWith(Constant.PARTITION_FILTER_HINT)) {
userConfiguredHintPartition.add(userConfiguredPartition.trim());
isIncludeHintPartition = true;
}else {
userConfiguredNormalPartition.add(userConfiguredPartition.trim());
}
}
}
userConfiguredPartitionClassification.setIncludeHintPartition(isIncludeHintPartition);
userConfiguredPartitionClassification.setUserConfiguredHintPartition(userConfiguredHintPartition);
userConfiguredPartitionClassification.setUserConfiguredNormalPartition(userConfiguredNormalPartition);
return userConfiguredPartitionClassification;
}
public static List<Pair<String, ColumnType>> parseColumns(
List<String> allNormalColumns, List<String> allPartitionColumns,
List<String> userConfiguredColumns) {
@@ -182,7 +222,7 @@ public final class OdpsUtil {
// warn: upper & lower case
for (String column : userConfiguredColumns) {
MutablePair<String, ColumnType> pair = new MutablePair<String, ColumnType>();
// if constant column
if (OdpsUtil.checkIfConstantColumn(column)) {
// remove first and last '
@@ -213,14 +253,14 @@ public final class OdpsUtil {
// not exist column
throw DataXException.asDataXException(
OdpsReaderErrorCode.ILLEGAL_VALUE,
String.format("源头表的列配置错误. 您所配置的列 [%s] 不存在.", column));
MESSAGE_SOURCE.message("odpsutil.5", column));
}
return parsededColumns;
}
private static int indexOfIgnoreCase(List<String> columnCollection,
String column) {
String column) {
int index = -1;
for (int i = 0; i < columnCollection.size(); i++) {
if (columnCollection.get(i).equalsIgnoreCase(column)) {
@@ -255,7 +295,7 @@ public final class OdpsUtil {
return tunnel.createDownloadSession(
projectName, tableName);
}
}, MAX_RETRY_TIME, 1000, true);
}, DataXCaseEnvUtil.getRetryTimes(MAX_RETRY_TIME), DataXCaseEnvUtil.getRetryInterval(1000), DataXCaseEnvUtil.getRetryExponential(true));
} catch (Exception e) {
throw DataXException.asDataXException(OdpsReaderErrorCode.CREATE_DOWNLOADSESSION_FAIL, e);
}
@@ -276,7 +316,7 @@ public final class OdpsUtil {
return tunnel.getDownloadSession(
projectName, tableName, sessionId);
}
}, MAX_RETRY_TIME ,1000, true);
}, DataXCaseEnvUtil.getRetryTimes(MAX_RETRY_TIME), DataXCaseEnvUtil.getRetryInterval(1000), DataXCaseEnvUtil.getRetryExponential(true));
} catch (Exception e) {
throw DataXException.asDataXException(OdpsReaderErrorCode.GET_DOWNLOADSESSION_FAIL, e);
}
@@ -299,7 +339,7 @@ public final class OdpsUtil {
return tunnel.createDownloadSession(
projectName, tableName, partitionSpec);
}
}, MAX_RETRY_TIME, 1000, true);
}, DataXCaseEnvUtil.getRetryTimes(MAX_RETRY_TIME), DataXCaseEnvUtil.getRetryInterval(1000), DataXCaseEnvUtil.getRetryExponential(true));
} catch (Exception e) {
throw DataXException.asDataXException(OdpsReaderErrorCode.CREATE_DOWNLOADSESSION_FAIL, e);
}
@@ -321,58 +361,152 @@ public final class OdpsUtil {
return tunnel.getDownloadSession(
projectName, tableName, partitionSpec, sessionId);
}
}, MAX_RETRY_TIME, 1000, true);
}, DataXCaseEnvUtil.getRetryTimes(MAX_RETRY_TIME), DataXCaseEnvUtil.getRetryInterval(1000), DataXCaseEnvUtil.getRetryExponential(true));
} catch (Exception e) {
throw DataXException.asDataXException(OdpsReaderErrorCode.GET_DOWNLOADSESSION_FAIL, e);
}
}
/**
* odpsreader采用的直接读取所有列的downloadSession
*/
public static RecordReader getRecordReader(final TableTunnel.DownloadSession downloadSession, final long start, final long count,
final boolean isCompress) {
final boolean isCompress) {
try {
return RetryUtil.executeWithRetry(new Callable<RecordReader>() {
@Override
public RecordReader call() throws Exception {
return downloadSession.openRecordReader(start, count, isCompress);
}
}, MAX_RETRY_TIME, 1000, true);
}, DataXCaseEnvUtil.getRetryTimes(MAX_RETRY_TIME), DataXCaseEnvUtil.getRetryInterval(1000), DataXCaseEnvUtil.getRetryExponential(true));
} catch (Exception e) {
throw DataXException.asDataXException(OdpsReaderErrorCode.OPEN_RECORD_READER_FAILED,
"open RecordReader失败. 请联系 ODPS 管理员处理.", e);
MESSAGE_SOURCE.message("odpsutil.6"), e);
}
}
/**
* odpsreader采用的指定读取某些列的downloadSession
*/
public static RecordReader getRecordReader(final TableTunnel.DownloadSession downloadSession, final long start, final long count,
final boolean isCompress, final List<Column> columns) {
try {
return RetryUtil.executeWithRetry(new Callable<RecordReader>() {
@Override
public RecordReader call() throws Exception {
return downloadSession.openRecordReader(start, count, isCompress, columns);
}
}, DataXCaseEnvUtil.getRetryTimes(MAX_RETRY_TIME), DataXCaseEnvUtil.getRetryInterval(1000), DataXCaseEnvUtil.getRetryExponential(true));
} catch (Exception e) {
throw DataXException.asDataXException(OdpsReaderErrorCode.OPEN_RECORD_READER_FAILED,
MESSAGE_SOURCE.message("odpsutil.6"), e);
}
}
/**
* table.reload() 方法抛出的 odps 异常 转化为更清晰的 datax 异常 抛出
*/
public static void throwDataXExceptionWhenReloadTable(Exception e, String tableName) {
if(e.getMessage() != null) {
if(e.getMessage().contains(OdpsExceptionMsg.ODPS_PROJECT_NOT_FOUNT)) {
if (e.getMessage() != null) {
if (e.getMessage().contains(OdpsExceptionMsg.ODPS_PROJECT_NOT_FOUNT)) {
throw DataXException.asDataXException(OdpsReaderErrorCode.ODPS_PROJECT_NOT_FOUNT,
String.format("加载 ODPS 源头表:%s 失败. " +
"请检查您配置的 ODPS 源头表的 [project] 是否正确.", tableName), e);
} else if(e.getMessage().contains(OdpsExceptionMsg.ODPS_TABLE_NOT_FOUNT)) {
MESSAGE_SOURCE.message("odpsutil.7", tableName), e);
} else if (e.getMessage().contains(OdpsExceptionMsg.ODPS_TABLE_NOT_FOUNT)) {
throw DataXException.asDataXException(OdpsReaderErrorCode.ODPS_TABLE_NOT_FOUNT,
String.format("加载 ODPS 源头表:%s 失败. " +
"请检查您配置的 ODPS 源头表的 [table] 是否正确.", tableName), e);
} else if(e.getMessage().contains(OdpsExceptionMsg.ODPS_ACCESS_KEY_ID_NOT_FOUND)) {
MESSAGE_SOURCE.message("odpsutil.8", tableName), e);
} else if (e.getMessage().contains(OdpsExceptionMsg.ODPS_ACCESS_KEY_ID_NOT_FOUND)) {
throw DataXException.asDataXException(OdpsReaderErrorCode.ODPS_ACCESS_KEY_ID_NOT_FOUND,
String.format("加载 ODPS 源头表:%s 失败. " +
"请检查您配置的 ODPS 源头表的 [accessId] [accessKey]是否正确.", tableName), e);
} else if(e.getMessage().contains(OdpsExceptionMsg.ODPS_ACCESS_KEY_INVALID)) {
MESSAGE_SOURCE.message("odpsutil.9", tableName), e);
} else if (e.getMessage().contains(OdpsExceptionMsg.ODPS_ACCESS_KEY_INVALID)) {
throw DataXException.asDataXException(OdpsReaderErrorCode.ODPS_ACCESS_KEY_INVALID,
String.format("加载 ODPS 源头表:%s 失败. " +
"请检查您配置的 ODPS 源头表的 [accessKey] 是否正确.", tableName), e);
} else if(e.getMessage().contains(OdpsExceptionMsg.ODPS_ACCESS_DENY)) {
MESSAGE_SOURCE.message("odpsutil.10", tableName), e);
} else if (e.getMessage().contains(OdpsExceptionMsg.ODPS_ACCESS_DENY)) {
throw DataXException.asDataXException(OdpsReaderErrorCode.ODPS_ACCESS_DENY,
String.format("加载 ODPS 源头表:%s 失败. " +
"请检查您配置的 ODPS 源头表的 [accessId] [accessKey] [project]是否匹配.", tableName), e);
MESSAGE_SOURCE.message("odpsutil.11", tableName), e);
}
}
throw DataXException.asDataXException(OdpsReaderErrorCode.ILLEGAL_VALUE,
String.format("加载 ODPS 源头表:%s 失败. " +
"请检查您配置的 ODPS 源头表的 project,table,accessId,accessKey,odpsServer等值.", tableName), e);
MESSAGE_SOURCE.message("odpsutil.12", tableName), e);
}
public static List<Column> getNormalColumns(List<Pair<String, ColumnType>> parsedColumns,
Map<String, TypeInfo> columnTypeMap) {
List<Column> userConfigNormalColumns = new ArrayList<Column>();
Set<String> columnNameSet = new HashSet<String>();
for (Pair<String, ColumnType> columnInfo : parsedColumns) {
if (columnInfo.getValue() == ColumnType.NORMAL) {
String columnName = columnInfo.getKey();
if (!columnNameSet.contains(columnName)) {
Column column = new Column(columnName, columnTypeMap.get(columnName));
userConfigNormalColumns.add(column);
columnNameSet.add(columnName);
}
}
}
return userConfigNormalColumns;
}
/**
* 执行odps preSql和postSql
*
* @param odps: odps client
* @param sql : 要执行的odps sql语句, 因为会有重试, 所以sql 必须为幂等的
* @param tag : "preSql" or "postSql"
*/
public static void runSqlTaskWithRetry(final Odps odps, final String sql, final String tag){
//重试次数
int retryTimes = 10;
//重试间隔(ms)
long sleepTimeInMilliSecond = 1000L;
try {
RetryUtil.executeWithRetry(new Callable<Void>() {
@Override
public Void call() throws Exception {
long beginTime = System.currentTimeMillis();
runSqlTask(odps, sql, tag);
long endIime = System.currentTimeMillis();
LOG.info(String.format("exectue odps sql: %s finished, cost time : %s ms",
sql, (endIime - beginTime)));
return null;
}
}, DataXCaseEnvUtil.getRetryTimes(retryTimes), DataXCaseEnvUtil.getRetryInterval(sleepTimeInMilliSecond), DataXCaseEnvUtil.getRetryExponential(true));
} catch (Exception e) {
String errMessage = String.format("Retry %s times to exectue sql :[%s] failed! Exception: %s",
retryTimes, e.getMessage());
throw DataXException.asDataXException(OdpsReaderErrorCode.RUN_SQL_ODPS_EXCEPTION, errMessage, e);
}
}
public static void runSqlTask(Odps odps, String sql, String tag) {
if (StringUtils.isBlank(sql)) {
return;
}
String taskName = String.format("datax_odpsreader_%s_%s", tag, UUID.randomUUID().toString().replace('-', '_'));
LOG.info("Try to start sqlTask:[{}] to run odps sql:[\n{}\n] .", taskName, sql);
Instance instance;
Instance.TaskStatus status;
try {
Map<String, String> hints = new HashMap<String, String>();
hints.put("odps.sql.submit.mode", "script");
instance = SQLTask.run(odps, odps.getDefaultProject(), sql, taskName, hints, null);
instance.waitForSuccess();
status = instance.getTaskStatus().get(taskName);
if (!Instance.TaskStatus.Status.SUCCESS.equals(status.getStatus())) {
throw DataXException.asDataXException(OdpsReaderErrorCode.RUN_SQL_FAILED,
MESSAGE_SOURCE.message("odpsutil.13", sql));
}
} catch (DataXException e) {
throw e;
} catch (Exception e) {
throw DataXException.asDataXException(OdpsReaderErrorCode.RUN_SQL_ODPS_EXCEPTION,
MESSAGE_SOURCE.message("odpsutil.14", sql), e);
}
}
}
odpsreader/src/main/java/com/alibaba/datax/plugin/reader/odpsreader/util/SqliteUtil.java
+103
View File
@@ -0,0 +1,103 @@
package com.alibaba.datax.plugin.reader.odpsreader.util;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.List;
import com.alibaba.datax.plugin.reader.odpsreader.Constant;
import com.aliyun.odps.Partition;
import com.aliyun.odps.Table;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class SqliteUtil {
private static final Logger LOGGER = LoggerFactory.getLogger(SqliteUtil.class);
private Connection connection = null;
private Statement stmt = null;
private String partitionName = "partitionName";
private String createSQLTemplate = "Create Table DataXODPSReaderPPR (" + partitionName +" String, %s)";
private String insertSQLTemplate = "Insert Into DataXODPSReaderPPR Values (%s)";
private String selectSQLTemplate = "Select * From DataXODPSReaderPPR Where %s";
public SqliteUtil() throws ClassNotFoundException, SQLException {
Class.forName("org.sqlite.JDBC");
this.connection = DriverManager.getConnection("jdbc:sqlite::memory:");
this.stmt = this.connection.createStatement();
}
public void loadAllPartitionsIntoSqlite(Table table, List<String> allOriginPartitions) throws SQLException {
List<String> partitionColumnList = new ArrayList<String>();
String partition = allOriginPartitions.get(0);
String[] partitionSpecs = partition.split(",");
List<String> partitionKeyList = new ArrayList<String>();
for (String partitionKeyValue : partitionSpecs) {
String partitionKey = partitionKeyValue.split("=")[0];
partitionColumnList.add(String.format("%s String", partitionKey));
partitionKeyList.add(partitionKey);
}
String createSQL = String.format(createSQLTemplate, StringUtils.join(partitionColumnList.toArray(), ","));
LOGGER.info(createSQL);
this.stmt.execute(createSQL);
insertAllOriginPartitionIntoSqlite(table, partitionKeyList);
}
/**
* 根据用户配置的过滤条件, 从sqlite中select出符合的partition列表
* @param userHintConfiguredPartitions
* @return
*/
public List<String> selectUserConfiguredPartition(List<String> userHintConfiguredPartitions) throws SQLException {
List<String> selectedPartitionsFromSqlite = new ArrayList<String>();
for (String partitionWhereConditions : userHintConfiguredPartitions) {
String selectUserConfiguredPartitionsSql = String.format(selectSQLTemplate,
StringUtils.remove(partitionWhereConditions, Constant.PARTITION_FILTER_HINT));
LOGGER.info(selectUserConfiguredPartitionsSql);
ResultSet rs = stmt.executeQuery(selectUserConfiguredPartitionsSql);
while (rs.next()) {
selectedPartitionsFromSqlite.add(getPartitionsValue(rs));
}
}
return selectedPartitionsFromSqlite;
}
private String getPartitionsValue (ResultSet rs) throws SQLException {
List<String> partitions = new ArrayList<String>();
ResultSetMetaData rsMetaData = rs.getMetaData();
Integer columnCounter = rs.getMetaData().getColumnCount();
for (int columnIndex = 2; columnIndex <= columnCounter; columnIndex++) {
partitions.add(String.format("%s=%s", rsMetaData.getColumnName(columnIndex), rs.getString(columnIndex)));
}
return StringUtils.join(partitions, ",");
}
/**
* 将odps table里所有partition值载入sqlite中
* @param table
* @param partitionKeyList
* @throws SQLException
*/
private void insertAllOriginPartitionIntoSqlite(Table table, List<String> partitionKeyList) throws SQLException {
List<Partition> partitions = table.getPartitions();
for (Partition partition : partitions){
List<String> partitionColumnValue = new ArrayList<String>();
partitionColumnValue.add("\""+partition.getPartitionSpec().toString()+"\"");
for (String partitionKey : partitionKeyList) {
partitionColumnValue.add("\""+partition.getPartitionSpec().get(partitionKey)+"\"");
}
String insertPartitionValueSql = String.format(insertSQLTemplate, StringUtils.join(partitionColumnValue, ","));
this.stmt.execute(insertPartitionValueSql);
}
}
}
odpsreader/src/main/java/com/alibaba/datax/plugin/reader/odpsreader/util/UserConfiguredPartitionClassification.java
+39
View File
@@ -0,0 +1,39 @@
package com.alibaba.datax.plugin.reader.odpsreader.util;
import java.util.List;
public class UserConfiguredPartitionClassification {
//包含/*query*/的partition, 例如: /*query*/ dt>=20170101 and dt<= 20170109
private List<String> userConfiguredHintPartition;
//不包含/*query*/的partition, 例如: dt=20170101 或者 dt=201701*
private List<String> userConfiguredNormalPartition;
//是否包含hint的partition
private boolean isIncludeHintPartition;
public List<String> getUserConfiguredHintPartition() {
return userConfiguredHintPartition;
}
public void setUserConfiguredHintPartition(List<String> userConfiguredHintPartition) {
this.userConfiguredHintPartition = userConfiguredHintPartition;
}
public List<String> getUserConfiguredNormalPartition() {
return userConfiguredNormalPartition;
}
public void setUserConfiguredNormalPartition(List<String> userConfiguredNormalPartition) {
this.userConfiguredNormalPartition = userConfiguredNormalPartition;
}
public boolean isIncludeHintPartition() {
return isIncludeHintPartition;
}
public void setIncludeHintPartition(boolean includeHintPartition) {
isIncludeHintPartition = includeHintPartition;
}
}
odpsreader/src/main/libs/bcprov-jdk15on-1.52.jar
BIN
View File
Binary file not shown.