This is an automated email from the ASF dual-hosted git repository.
wanghailin pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new b592014766 [Feature][HbaseSink]support array data. (#6100)
b592014766 is described below
commit b5920147667d7a756d16624c4913328f16b0bae5
Author: lightzhao <[email protected]>
AuthorDate: Tue Jan 2 12:03:39 2024 +0800
[Feature][HbaseSink]support array data. (#6100)
---
.../seatunnel/hbase/sink/HbaseSinkWriter.java | 4 ++
.../seatunnel/e2e/connector/hbase/HbaseIT.java | 29 +++++++++++
.../src/test/resources/fake-to-hbase-array.conf | 59 ++++++++++++++++++++++
3 files changed, 92 insertions(+)
diff --git
a/seatunnel-connectors-v2/connector-hbase/src/main/java/org/apache/seatunnel/connectors/seatunnel/hbase/sink/HbaseSinkWriter.java
b/seatunnel-connectors-v2/connector-hbase/src/main/java/org/apache/seatunnel/connectors/seatunnel/hbase/sink/HbaseSinkWriter.java
index 5df5e1fd75..0455245c67 100644
---
a/seatunnel-connectors-v2/connector-hbase/src/main/java/org/apache/seatunnel/connectors/seatunnel/hbase/sink/HbaseSinkWriter.java
+++
b/seatunnel-connectors-v2/connector-hbase/src/main/java/org/apache/seatunnel/connectors/seatunnel/hbase/sink/HbaseSinkWriter.java
@@ -179,6 +179,10 @@ public class HbaseSinkWriter extends
AbstractSinkWriter<SeaTunnelRow, Void> {
return Bytes.toBytes((Double) field);
case BOOLEAN:
return Bytes.toBytes((Boolean) field);
+ case ARRAY:
+ String arrayAsString =
field.toString().replaceAll("\\[|\\]|\\s", "");
+ return arrayAsString.getBytes(
+
Charset.forName(hbaseParameters.getEnCoding().toString()));
case STRING:
return field.toString()
.getBytes(Charset.forName(hbaseParameters.getEnCoding().toString()));
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-hbase-e2e/src/test/java/org/apache/seatunnel/e2e/connector/hbase/HbaseIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-hbase-e2e/src/test/java/org/apache/seatunnel/e2e/connector/hbase/HbaseIT.java
index 19b786b950..4c7a9587ea 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-hbase-e2e/src/test/java/org/apache/seatunnel/e2e/connector/hbase/HbaseIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-hbase-e2e/src/test/java/org/apache/seatunnel/e2e/connector/hbase/HbaseIT.java
@@ -22,6 +22,8 @@ import org.apache.seatunnel.e2e.common.TestSuiteBase;
import org.apache.seatunnel.e2e.common.container.TestContainer;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.Cell;
+import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Admin;
@@ -36,6 +38,7 @@ import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.client.TableDescriptor;
import org.apache.hadoop.hbase.client.TableDescriptorBuilder;
import org.apache.hadoop.hbase.io.compress.Compression;
+import org.apache.hadoop.hbase.util.Bytes;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.Assertions;
@@ -143,4 +146,30 @@ public class HbaseIT extends TestSuiteBase implements
TestResource {
}
Assertions.assertEquals(results.size(), 5);
}
+
+ @TestTemplate
+ public void testHbaseSinkWithArray(TestContainer container)
+ throws IOException, InterruptedException {
+ Container.ExecResult execResult =
container.executeJob("/fake-to-hbase-array.conf");
+ Assertions.assertEquals(0, execResult.getExitCode());
+ Table hbaseTable = hbaseConnection.getTable(table);
+ Scan scan = new Scan();
+ ArrayList<Result> results = new ArrayList<>();
+ ResultScanner scanner = hbaseTable.getScanner(scan);
+ for (Result result : scanner) {
+ String rowKey = Bytes.toString(result.getRow());
+ for (Cell cell : result.listCells()) {
+ String columnName =
Bytes.toString(CellUtil.cloneQualifier(cell));
+ String value = Bytes.toString(CellUtil.cloneValue(cell));
+ if ("A".equals(rowKey) &&
"info:c_array_string".equals(columnName)) {
+ Assertions.assertEquals(value, "\"a\",\"b\",\"c\"");
+ }
+ if ("B".equals(rowKey) &&
"info:c_array_int".equals(columnName)) {
+ Assertions.assertEquals(value, "4,5,6");
+ }
+ }
+ results.add(result);
+ }
+ Assertions.assertEquals(results.size(), 3);
+ }
}
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-hbase-e2e/src/test/resources/fake-to-hbase-array.conf
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-hbase-e2e/src/test/resources/fake-to-hbase-array.conf
new file mode 100644
index 0000000000..5cf1896ea2
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-hbase-e2e/src/test/resources/fake-to-hbase-array.conf
@@ -0,0 +1,59 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You 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.
+#
+
+env {
+ execution.parallelism = 1
+ job.mode = "BATCH"
+}
+
+source {
+ FakeSource {
+ schema = {
+ fields {
+ name = string
+ score = int
+ c_array_string = "array<string>"
+ c_array_int = "array<int>"
+ }
+ }
+ rows = [
+ {
+ kind = INSERT
+ fields = ["A", 100,["a","b","c"],[1,2,3]]
+ },
+ {
+ kind = INSERT
+ fields = ["B", 200,["d","e","f"],[4,5,6]]
+ },
+ {
+ kind = INSERT
+ fields = ["C", 300,["g","h","k"],[7,8,9]]
+ }
+ ]
+ }
+}
+
+sink {
+ Hbase {
+ zookeeper_quorum = "hbase-e2e:2181"
+ table = "seatunnel_test"
+ rowkey_column = ["name"]
+ family_name {
+ all_columns = info
+ }
+ }
+}
\ No newline at end of file
