Skip to content

Commit

Permalink
Merge pull request alibaba#1606 from hadoop835/master
Browse files Browse the repository at this point in the history 
修复 Phoenix4版本同步数据添加where条件
  • Loading branch information
@jtchen-study
jtchen-study authored Sep 7, 2023
2 parents 4acbe4d + cf2b141 commit 8db76dd
Show file tree
Hide file tree
Showing 5 changed files with 104 additions and 24 deletions.
42 changes: 37 additions & 5 deletions hbase11xsqlreader/doc/hbase11xsqlreader.md
View file
Original file line number Diff line number Diff line change
Expand Up @@ -60,12 +60,16 @@ hbase11xsqlreader插件实现了从Phoenix(HBase SQL)读取数据。在底层实
//填写连接Phoenix的hbase集群zk地址
"hbaseConfig": {
"hbase.zookeeper.quorum": "hb-proxy-xxx-002.hbase.rds.aliyuncs.com,hb-proxy-xxx-001.hbase.rds.aliyuncs.com,hb-proxy-xxx-003.hbase.rds.aliyuncs.com"
},
},
//填写要读取的phoenix的命名空间
"schema": "TAG",
//填写要读取的phoenix的表名
"table": "US_POPULATION",
//填写要读取的列名，不填读取所有列
"column": [
]
],
//查询条件
"where": "id="
}
},
"writer": {
Expand All @@ -92,11 +96,18 @@ hbase11xsqlreader插件实现了从Phoenix(HBase SQL)读取数据。在底层实

* 必选：是 <br />

* 默认值：无 <br />
* **schema**

* 描述：编写Phoenix中的namespace，该值设置为''

* 必选：是 <br />

* 默认值：无 <br />

* **table**

* 描述：编写Phoenix中的表名,如果有namespace，该值设置为'namespace.tablename'
* 描述：编写Phoenix中的表名，该值设置为'tablename'

* 必选：是 <br />

Expand All @@ -109,7 +120,13 @@ hbase11xsqlreader插件实现了从Phoenix(HBase SQL)读取数据。在底层实
* 必选：是 <br />

* 默认值：无 <br />
* **where**

* 描述：填写需要从phoenix表中读取条件判断。

* 可选：是 <br />

* 默认值：无 <br />

### 3.3 类型转换

Expand Down Expand Up @@ -172,11 +189,14 @@ hbase11xsqlreader插件实现了从Phoenix(HBase SQL)读取数据。在底层实
"hbaseConfig": {
"hbase.zookeeper.quorum": "hb-proxy-xxx-002.hbase.rds.aliyuncs.com,hb-proxy-xxx-001.hbase.rds.aliyuncs.com,hb-proxy-xxx-003.hbase.rds.aliyuncs.com"
},
"schema": "TAG",
//填写要读取的phoenix的表名
"table": "US_POPULATION",
//填写要读取的列名，不填读取所有列
"column": [
]
],
//查询条件
"where": "id="
}
},
"writer": {
Expand Down Expand Up @@ -204,7 +224,13 @@ hbase11xsqlreader插件实现了从Phoenix(HBase SQL)读取数据。在底层实
* 必选：是 <br />

* 默认值：无 <br />

* **schema**

* 描述：编写Phoenix中的namespace，该值设置为''

* 必选：是 <br />

* 默认值：无 <br />
* **table**

* 描述：编写Phoenix中的表名,如果有namespace，该值设置为'namespace.tablename'
Expand All @@ -220,7 +246,13 @@ hbase11xsqlreader插件实现了从Phoenix(HBase SQL)读取数据。在底层实
* 必选：是 <br />

* 默认值：无 <br />
* **where**

* 描述：填写需要从phoenix表中读取条件判断。

* 可选：是 <br />

* 默认值：无 <br />

### 3.3 类型转换

Expand Down
22 changes: 14 additions & 8 deletions ...eader/src/main/java/com/alibaba/datax/plugin/reader/hbase11xsqlreader/HbaseSQLHelper.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -26,9 +26,7 @@
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.*;


public class HbaseSQLHelper {
Expand All @@ -50,11 +48,15 @@ public static org.apache.hadoop.conf.Configuration generatePhoenixConf(HbaseSQLR
String zkUrl = readerConfig.getZkUrl();

PhoenixConfigurationUtil.setInputClass(conf, PhoenixRecordWritable.class);
PhoenixConfigurationUtil.setInputTableName(conf, table);

PhoenixConfigurationUtil.setInputTableName(conf, readerConfig.getSchema()+"."+table);

if (!columns.isEmpty()) {
PhoenixConfigurationUtil.setSelectColumnNames(conf, columns.toArray(new String[columns.size()]));
}
if(Objects.nonNull(readerConfig.getWhere())){
PhoenixConfigurationUtil.setInputTableConditions(conf,readerConfig.getWhere());
}
PhoenixEmbeddedDriver.ConnectionInfo info = null;
try {
info = PhoenixEmbeddedDriver.ConnectionInfo.create(zkUrl);
Expand All @@ -67,15 +69,19 @@ public static org.apache.hadoop.conf.Configuration generatePhoenixConf(HbaseSQLR
conf.setInt(HConstants.ZOOKEEPER_CLIENT_PORT, info.getPort());
if (info.getRootNode() != null)
conf.set(HConstants.ZOOKEEPER_ZNODE_PARENT, info.getRootNode());
conf.set(Key.NAME_SPACE_MAPPING_ENABLED,"true");
conf.set(Key.SYSTEM_TABLES_TO_NAMESPACE,"true");
return conf;
}

public static List<String> getPColumnNames(String connectionString, String tableName) throws SQLException {
Connection con =
DriverManager.getConnection(connectionString);
public static List<String> getPColumnNames(String connectionString, String tableName,String schema) throws SQLException {
Properties pro = new Properties();
pro.put(Key.NAME_SPACE_MAPPING_ENABLED, true);
pro.put(Key.SYSTEM_TABLES_TO_NAMESPACE, true);
Connection con = DriverManager.getConnection(connectionString,pro);
PhoenixConnection phoenixConnection = con.unwrap(PhoenixConnection.class);
MetaDataClient metaDataClient = new MetaDataClient(phoenixConnection);
PTable table = metaDataClient.updateCache("", tableName).getTable();
PTable table = metaDataClient.updateCache(schema, tableName).getTable();
List<String> columnNames = new ArrayList<String>();
for (PColumn pColumn : table.getColumns()) {
if (!pColumn.getName().getString().equals(SaltingUtil.SALTING_COLUMN_NAME))
Expand Down
38 changes: 33 additions & 5 deletions ...src/main/java/com/alibaba/datax/plugin/reader/hbase11xsqlreader/HbaseSQLReaderConfig.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@

import java.sql.SQLException;
import java.util.List;
import java.util.StringJoiner;

public class HbaseSQLReaderConfig {
private final static Logger LOG = LoggerFactory.getLogger(HbaseSQLReaderConfig.class);
Expand All @@ -27,6 +28,9 @@ public String getZkUrl() {
private String tableName;
private List<String> columns; // 目的表的所有列的列名，包括主键和非主键，不包括时间列

private String where;//条件

private String schema;//
/**
* @return 获取原始的datax配置
*/
Expand Down Expand Up @@ -96,22 +100,27 @@ private static void parseClusterConfig(HbaseSQLReaderConfig cfg, Configuration d
}
String zkQuorum = zkCfg.getFirst();
String znode = zkCfg.getSecond();

if (zkQuorum == null || zkQuorum.isEmpty()) {
throw DataXException.asDataXException(
HbaseSQLReaderErrorCode.ILLEGAL_VALUE, "HBase的hbase.zookeeper.quorum配置不能为空" );
}
// 生成sql使用的连接字符串， 格式： jdbc:hbase:zk_quorum:2181:/znode_parent
cfg.connectionString = "jdbc:phoenix:" + zkQuorum;
cfg.zkUrl = zkQuorum + ":2181";
StringBuilder connectionString=new StringBuilder("jdbc:phoenix:");
connectionString.append(zkQuorum);
cfg.connectionString = connectionString.toString();
StringBuilder zkUrl =new StringBuilder(zkQuorum);
cfg.zkUrl = zkUrl.append(":2181").toString();
if (!znode.isEmpty()) {
cfg.connectionString += cfg.connectionString + ":" + znode;
cfg.zkUrl += cfg.zkUrl + ":" + znode;
cfg.connectionString = connectionString.append(":").append(znode).toString();
cfg.zkUrl=zkUrl.append(":").append(znode).toString();
}
}

private static void parseTableConfig(HbaseSQLReaderConfig cfg, Configuration dataxCfg) {
// 解析并检查表名
cfg.tableName = dataxCfg.getString(Key.TABLE);
cfg.schema = dataxCfg.getString(Key.SCHEMA);
if (cfg.tableName == null || cfg.tableName.isEmpty()) {
throw DataXException.asDataXException(
HbaseSQLReaderErrorCode.ILLEGAL_VALUE, "HBase的tableName配置不能为空,请检查并修改配置." );
Expand All @@ -124,13 +133,14 @@ private static void parseTableConfig(HbaseSQLReaderConfig cfg, Configuration dat
HbaseSQLReaderErrorCode.ILLEGAL_VALUE, "您配置的tableName含有非法字符{0}，请检查您的配置.");
} else if (cfg.columns.isEmpty()) {
try {
cfg.columns = HbaseSQLHelper.getPColumnNames(cfg.connectionString, cfg.tableName);
cfg.columns = HbaseSQLHelper.getPColumnNames(cfg.connectionString, cfg.tableName,cfg.schema);
dataxCfg.set(Key.COLUMN, cfg.columns);
} catch (SQLException e) {
throw DataXException.asDataXException(
HbaseSQLReaderErrorCode.GET_PHOENIX_COLUMN_ERROR, "HBase的columns配置不能为空,请添加目标表的列名配置." + e.getMessage(), e);
}
}
cfg.where=dataxCfg.getString(Key.WHERE);
}

@Override
Expand All @@ -151,6 +161,8 @@ public String toString() {
ret.append(",");
}
ret.setLength(ret.length() - 1);
ret.append("[where=]").append(getWhere());
ret.append("[schema=]").append(getSchema());
ret.append("\n");

return ret.toString();
Expand All @@ -161,4 +173,20 @@ public String toString() {
*/
private HbaseSQLReaderConfig() {
}

public String getWhere() {
return where;
}

public void setWhere(String where) {
this.where = where;
}

public String getSchema() {
return schema;
}

public void setSchema(String schema) {
this.schema = schema;
}
}
13 changes: 7 additions & 6 deletions ...r/src/main/java/com/alibaba/datax/plugin/reader/hbase11xsqlreader/HbaseSQLReaderTask.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -19,10 +19,8 @@
import java.io.IOException;
import java.math.BigDecimal;
import java.sql.*;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.sql.Date;
import java.util.*;

/**
* Created by admin on 1/3/18.
Expand All @@ -42,11 +40,14 @@ public HbaseSQLReaderTask(Configuration config) {
}

private void getPColumns() throws SQLException {
Properties pro = new Properties();
pro.put(Key.NAME_SPACE_MAPPING_ENABLED, true);
pro.put(Key.SYSTEM_TABLES_TO_NAMESPACE, true);
Connection con =
DriverManager.getConnection(this.readerConfig.getConnectionString());
DriverManager.getConnection(this.readerConfig.getConnectionString(),pro);
PhoenixConnection phoenixConnection = con.unwrap(PhoenixConnection.class);
MetaDataClient metaDataClient = new MetaDataClient(phoenixConnection);
PTable table = metaDataClient.updateCache("", this.readerConfig.getTableName()).getTable();
PTable table = metaDataClient.updateCache(this.readerConfig.getSchema(), this.readerConfig.getTableName()).getTable();
List<String> columnNames = this.readerConfig.getColumns();
for (PColumn pColumn : table.getColumns()) {
if (columnNames.contains(pColumn.getName().getString())) {
Expand Down
13 changes: 13 additions & 0 deletions hbase11xsqlreader/src/main/java/com/alibaba/datax/plugin/reader/hbase11xsqlreader/Key.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -24,5 +24,18 @@ public final class Key {
* 【必选】列配置
*/
public final static String COLUMN = "column";
/**
*
*/
public static final String WHERE = "where";

/**
* 【可选】Phoenix表所属schema，默认为空
*/
public static final String SCHEMA = "schema";

public static final String NAME_SPACE_MAPPING_ENABLED = "phoenix.schema.isNamespaceMappingEnabled";

public static final String SYSTEM_TABLES_TO_NAMESPACE = "phoenix.schema.mapSystemTablesToNamespace";

}

0 comments on commit 8db76dd

Please sign in to comment.