DataX二次开发——新增HiveReader插件

一、研发背景

    DataX官方开源的版本支持HDFS文件的读写，并没有支持基于JDBC的Hive数据读写，很多时候一些数据同步不太方便，比如在读取Hive之前先执行一些sql、读取一些Hive的视图数据、或者在数据同步时执行一段固定的SQL，将SQL执行结果写入下游等各种场景，实际上还是需要Hive插件来支持。而在实际工作中，我们也遇到了类似的一些情况需要二次开发DataX以支持此类场景。本插件已在生产环境稳定运行一年有余，现分享给大家，如有问题也可联系我（qq：1821088755）。
二、HiveReader插件介绍

    hivereader插件比较简单，共有三个类，两个配置文件。其中：

• HiveReader：实现DataX框架核心方法，是具体逻辑。
  
• HiveReaderErrorCode：继承了DataX框架的ErrorCode类，是用于统一异常处理DataXException类中调用，具体是新增了一个枚举值。
  
• HiveConnByKerberos：是在检测到Hive具备Kerberos认证要求时，进行认证的工具类。
  
• plugin.json：DataX插件固定的配置文件，用于指定插件的入口类。
  
• plugin_job_template.json：二次开发插件，一般需要提供一下具体的使用方式，此json文件即为HiveReader插件的配置方式说明。
  

 
   2.1 HiveReader类

    首先是HiveReader类，需要注意的是一些常量或枚举值，需要自行添加，其中DataBaseType枚举类中，需要新增Hive枚举项并添加Hive的驱动类全路径，具体见注释，另外就是Kerberos认证相关的几个配置，一个是keytab的路径，一个是krb5.conf的路径，另外一个是principle的值。

1. package com.alibaba.datax.plugin.reader.hivereader;
3. import com.alibaba.datax.common.base.Key;
4. import com.alibaba.datax.common.plugin.RecordSender;
5. import com.alibaba.datax.common.spi.Reader;
6. import com.alibaba.datax.common.util.Configuration;
7. import com.alibaba.datax.rdbms.reader.CommonRdbmsReader;
8. import com.alibaba.datax.rdbms.util.DataBaseType;
9. import lombok.extern.slf4j.Slf4j;
10. import org.apache.hadoop.security.authentication.util.KerberosName;
12. import java.lang.reflect.Field;
13. import java.util.List;
15. import static com.alibaba.datax.common.base.Constant.DEFAULT_FETCH_SIZE;//2048，可根据条件自己取值
16. import static com.alibaba.datax.common.base.Key.FETCH_SIZE; // 参数名："fetchSize"
18. @Slf4j
19. public class HiveReader
20.         extends Reader
21. {
23.     //此处需现在com.sinosig.plumber.rdbms.util.DataBaseType枚举类中添加Hive类型，内容为：Hive("hive2", "org.apache.hive.jdbc.HiveDriver"),
24.     private static final DataBaseType DATABASE_TYPE = DataBaseType.Hive;
26.     public static class Job
27.             extends Reader.Job
28.     {
30.         private Configuration originalConfig = null;
31.         private CommonRdbmsReader.Job commonRdbmsReaderJob;
33.         @Override
34.         public void init()
35.         {
36.             this.originalConfig = getPluginJobConf();
39.             Boolean haveKerberos = this.originalConfig.getBool(Key.HAVE_KERBEROS, false);
40.             if (haveKerberos) {
41.                 log.info("检测到kerberos认证，正在进行认证");
42.                 org.apache.hadoop.conf.Configuration hadoopConf = new org.apache.hadoop.conf.Configuration();
43.                 String kerberosKeytabFilePath =  this.originalConfig.getString(Key.KERBEROS_KEYTAB_FILE_PATH);
44.                 String kerberosPrincipal =  this.originalConfig.getString(Key.KERBEROS_PRINCIPAL);
45.                 String krb5Path =  this.originalConfig.getString(Key.KRB5_CONF_FILE_PATH);
47.                 hadoopConf.set("hadoop.security.authentication", "kerberos");
48.                 hadoopConf.set("hive.security.authentication", "kerberos");
49.                 hadoopConf.set("hadoop.security.authorization", "true");
50.                 System.setProperty("java.security.krb5.conf",krb5Path);
51.                 refreshConfig();
52.                 HiveConnByKerberos.kerberosAuthentication(kerberosPrincipal, kerberosKeytabFilePath, hadoopConf,krb5Path);
53.             }
54.             this.commonRdbmsReaderJob = new CommonRdbmsReader.Job(DATABASE_TYPE);
55.             this.originalConfig = commonRdbmsReaderJob.init(originalConfig);
56.         }
59.         @Override
60.         public void preCheck()
61.         {
62.             this.commonRdbmsReaderJob.preCheck(originalConfig, DATABASE_TYPE);
63.         }
65.         @Override
66.         public List<Configuration> split(int adviceNumber)
67.         {
68.             return this.commonRdbmsReaderJob.split(originalConfig, adviceNumber);
69.         }
71.         @Override
72.         public void post()
73.         {
74.             this.commonRdbmsReaderJob.post(originalConfig);
75.         }
77.         @Override
78.         public void destroy()
79.         {
80.             this.commonRdbmsReaderJob.destroy(originalConfig);
81.         }
84.     }
86.     public static class Task
87.             extends Reader.Task
88.     {
90.         private Configuration readerSliceConfig;
91.         private CommonRdbmsReader.Task commonRdbmsReaderTask;
93.         @Override
94.         public void init()
95.         {
96.             this.readerSliceConfig = getPluginJobConf();
97.             this.commonRdbmsReaderTask = new CommonRdbmsReader.Task(DATABASE_TYPE, getTaskGroupId(), getTaskId());
98.             this.commonRdbmsReaderTask.init(this.readerSliceConfig);
99.         }
101.         @Override
102.         public void startRead(RecordSender recordSender)
103.         {
104.             int fetchSize = this.readerSliceConfig.getInt(FETCH_SIZE, DEFAULT_FETCH_SIZE);
106.             this.commonRdbmsReaderTask.startRead(readerSliceConfig, recordSender, getTaskPluginCollector(), fetchSize);
107.         }
109.         @Override
110.         public void post()
111.         {
112.             this.commonRdbmsReaderTask.post(readerSliceConfig);
113.         }
115.         @Override
116.         public void destroy()
117.         {
118.             this.commonRdbmsReaderTask.destroy(readerSliceConfig);
119.         }
120.     }
121.     /** 刷新krb内容信息 */
122.     public static void refreshConfig() {
123.         try {
124.             sun.security.krb5.Config.refresh();
125.             Field defaultRealmField = KerberosName.class.getDeclaredField("defaultRealm");
126.             defaultRealmField.setAccessible(true);
127.             defaultRealmField.set(
128.                     null,
129.                     org.apache.hadoop.security.authentication.util.KerberosUtil.getDefaultRealm());
130.             // reload java.security.auth.login.config
131.             javax.security.auth.login.Configuration.setConfiguration(null);
132.         } catch (Exception e) {
133.             log.warn(
134.                     "resetting default realm failed, current default realm will still be used.", e);
135.         }
136.     }
137. }

复制代码

 2.2 HiveConnByKerberos类

    HiveConnByKerberos类比较简单，是一个通用的Kerberos认证的接口。

1. package com.alibaba.datax.plugin.reader.hivereader;
3. import com.alibaba.datax.common.exception.PlumberException;
4. import lombok.extern.slf4j.Slf4j;
5. import org.apache.commons.lang3.StringUtils;
6. import org.apache.hadoop.security.UserGroupInformation;
8. @Slf4j
9. public class HiveConnByKerberos {
10.     public static void kerberosAuthentication(String kerberosPrincipal, String kerberosKeytabFilePath, org.apache.hadoop.conf.Configuration hadoopConf,String krb5conf) {
11.         System.setProperty("java.security.krb5.conf",krb5conf);
12.         if (StringUtils.isNotBlank(kerberosPrincipal) && StringUtils.isNotBlank(kerberosKeytabFilePath)) {
13.             UserGroupInformation.setConfiguration(hadoopConf);
14.             try {
15.                 UserGroupInformation.loginUserFromKeytab(kerberosPrincipal, kerberosKeytabFilePath);
16.             }
17.             catch (Exception e) {
19.                 log.error("kerberos认证失败");
20.                 String message = String.format("kerberos认证失败，请检查 " +
21.                                 "kerberosKeytabFilePath[%s] 和 kerberosPrincipal[%s]",
22.                         kerberosKeytabFilePath, kerberosPrincipal);
23.                 e.printStackTrace();
24.                 throw DataXException.asDataXException(HiveReaderErrorCode.KERBEROS_LOGIN_ERROR, message, e);
25.             }
26.         }
27.     }
28. }

复制代码

 
2.3 HiveReaderErrorCode类

    HiveReaderErrorCode类，主要就是集成ErrorCode类，并添加一个枚举项，这块可直接在ErrorCode类添加，也可使用此类，为固定写法。

1. package com.alibaba.datax.plugin.reader.hivereader;
3. import com.alibaba.datax.common.spi.ErrorCode;
5. public enum HiveReaderErrorCode
6.         implements ErrorCode
7. {
8.     KERBEROS_LOGIN_ERROR("HiveReader-13", "KERBEROS认证失败");
10.     private final String code;
11.     private final String description;
13.     HiveReaderErrorCode(String code, String description)
14.     {
15.         this.code = code;
16.         this.description = description;
17.     }
19.     @Override
20.     public String getCode()
21.     {
22.         return this.code;
23.     }
25.     @Override
26.     public String getDescription()
27.     {
28.         return this.description;
29.     }
31.     @Override
32.     public String toString()
33.     {
34.         return String.format("Code:[%s], Description:[%s]. ", this.code, this.description);
35.     }
36. }

复制代码

2.4 plugin.json文件

1. {
2.   "name": "hivereader",
3.   "class": "com.alibaba.datax.plugin.reader.hivereader.HiveReader",
4.   "description": "Retrieve data from Hive via jdbc",
5.   "developer": "wxm"
6. }

复制代码

2.5 plugin_job_template.json文件

    这块需要注意的一个问题是，如果Kerberos认证的Hive连接URL有两种方式，如果是基于zookeeper的方式，则需保证运行DataX服务的节点与zookeeper节点网络是打通的，并且一定不要忘记写上具体的Hive库名。

1. {
2.   "name": "hivereader",
3.   "parameter": {
4.     "column": [
5.       "*"
6.     ],
7.     "username": "hive",
8.     "password": "",<br>    "preSql":"show databases;",
9.     "connection": [
10.       {
11.         "jdbcUrl": [
12.           "jdbc:hive2://localhost:10000/default;principal=hive/_HOST@EXAMPLE.COM"
13.         ],
14.         "table": [
15.           "hive_reader"
16.         ]
17.       }
18.     ],
19.     "where": "logdate='20211013'" ,
20.     "haveKerberos": true,
21.     "kerberosKeytabFilePath": "/etc/security/keytabs/hive.headless.keytab",
22.     "kerberosPrincipal": "hive@EXAMPLE.COM"
23.   }
24. }

复制代码

 

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！