You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@phoenix.apache.org by ja...@apache.org on 2014/03/16 19:22:04 UTC
[1/2] PHOENIX-129 Improve MapReduce-based import (GabrielReid)
Repository: incubator-phoenix
Updated Branches:
refs/heads/3.0 6451b1df4 -> ed9f58bae
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/ed9f58ba/phoenix-core/src/test/java/org/apache/phoenix/mapreduce/CsvBulkLoadToolTest.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/test/java/org/apache/phoenix/mapreduce/CsvBulkLoadToolTest.java b/phoenix-core/src/test/java/org/apache/phoenix/mapreduce/CsvBulkLoadToolTest.java
new file mode 100644
index 0000000..31fc71c
--- /dev/null
+++ b/phoenix-core/src/test/java/org/apache/phoenix/mapreduce/CsvBulkLoadToolTest.java
@@ -0,0 +1,80 @@
+/*
+ * 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.
+ */
+package org.apache.phoenix.mapreduce;
+
+import org.apache.commons.cli.CommandLine;
+import org.junit.Before;
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+
+public class CsvBulkLoadToolTest {
+
+ private CsvBulkLoadTool bulkLoadTool;
+
+ @Before
+ public void setUp() {
+ bulkLoadTool = new CsvBulkLoadTool();
+ }
+
+ @Test
+ public void testParseOptions() {
+ CommandLine cmdLine = bulkLoadTool.parseOptions(new String[] { "--input", "/input",
+ "--table", "mytable" });
+
+ assertEquals("mytable", cmdLine.getOptionValue(CsvBulkLoadTool.TABLE_NAME_OPT.getOpt()));
+ assertEquals("/input", cmdLine.getOptionValue(CsvBulkLoadTool.INPUT_PATH_OPT.getOpt()));
+ }
+
+ @Test(expected=IllegalStateException.class)
+ public void testParseOptions_ExtraArguments() {
+ bulkLoadTool.parseOptions(new String[] { "--input", "/input",
+ "--table", "mytable", "these", "shouldnt", "be", "here" });
+ }
+
+ @Test(expected=IllegalStateException.class)
+ public void testParseOptions_NoInput() {
+ bulkLoadTool.parseOptions(new String[] { "--table", "mytable" });
+ }
+
+ @Test(expected=IllegalStateException.class)
+ public void testParseOptions_NoTable() {
+ bulkLoadTool.parseOptions(new String[] { "--input", "/input" });
+ }
+
+ @Test
+ public void testGetQualifiedTableName() {
+ assertEquals("MYSCHEMA.MYTABLE", CsvBulkLoadTool.getQualifiedTableName("mySchema", "myTable"));
+ }
+
+ @Test
+ public void testGetQualifiedTableName_NullSchema() {
+ assertEquals("MYTABLE", CsvBulkLoadTool.getQualifiedTableName(null, "myTable"));
+ }
+
+ @Test
+ public void testGetJdbcUrl_WithQuorumSupplied() {
+ assertEquals("jdbc:phoenix:myzkhost:2181", bulkLoadTool.getJdbcUrl("myzkhost:2181"));
+ }
+
+ @Test
+ public void testGetJdbcUrl_NoQuorumSupplied() {
+ assertEquals("jdbc:phoenix:localhost:2181", bulkLoadTool.getJdbcUrl(null));
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/ed9f58ba/phoenix-core/src/test/java/org/apache/phoenix/mapreduce/CsvToKeyValueMapperTest.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/test/java/org/apache/phoenix/mapreduce/CsvToKeyValueMapperTest.java b/phoenix-core/src/test/java/org/apache/phoenix/mapreduce/CsvToKeyValueMapperTest.java
new file mode 100644
index 0000000..96157a6
--- /dev/null
+++ b/phoenix-core/src/test/java/org/apache/phoenix/mapreduce/CsvToKeyValueMapperTest.java
@@ -0,0 +1,132 @@
+/*
+ * 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.
+ */
+package org.apache.phoenix.mapreduce;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Lists;
+import org.apache.commons.csv.CSVRecord;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.HConstants;
+import org.apache.hadoop.hbase.KeyValue;
+import org.apache.phoenix.schema.PDataType;
+import org.apache.phoenix.util.ColumnInfo;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.util.List;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+public class CsvToKeyValueMapperTest {
+
+ @Test
+ public void testCsvLineParser() throws IOException {
+ CsvToKeyValueMapper.CsvLineParser lineParser = new CsvToKeyValueMapper.CsvLineParser(';');
+ CSVRecord parsed = lineParser.parse("one;two");
+
+ assertEquals("one", parsed.get(0));
+ assertEquals("two", parsed.get(1));
+ assertTrue(parsed.isConsistent());
+ assertEquals(1, parsed.getRecordNumber());
+ }
+
+
+ @Test
+ public void testBuildColumnInfoList() {
+ List<ColumnInfo> columnInfoList = ImmutableList.of(
+ new ColumnInfo("idCol", PDataType.INTEGER.getSqlType()),
+ new ColumnInfo("unsignedIntCol", PDataType.UNSIGNED_INT.getSqlType()),
+ new ColumnInfo("stringArrayCol", PDataType.INTEGER_ARRAY.getSqlType()));
+
+ Configuration conf = new Configuration();
+ CsvToKeyValueMapper.configureColumnInfoList(conf, columnInfoList);
+ List<ColumnInfo> fromConfig = CsvToKeyValueMapper.buildColumnInfoList(conf);
+
+ assertEquals(columnInfoList, fromConfig);
+ }
+
+ @Test
+ public void testBuildColumnInfoList_ContainingNulls() {
+ // A null value in the column info list means "skip that column in the input"
+ List<ColumnInfo> columnInfoListWithNull = Lists.newArrayList(
+ new ColumnInfo("idCol", PDataType.INTEGER.getSqlType()),
+ null,
+ new ColumnInfo("unsignedIntCol", PDataType.UNSIGNED_INT.getSqlType()),
+ new ColumnInfo("stringArrayCol", PDataType.INTEGER_ARRAY.getSqlType()));
+
+ Configuration conf = new Configuration();
+ CsvToKeyValueMapper.configureColumnInfoList(conf, columnInfoListWithNull);
+ List<ColumnInfo> fromConfig = CsvToKeyValueMapper.buildColumnInfoList(conf);
+
+ assertEquals(columnInfoListWithNull, fromConfig);
+ }
+
+ @Test
+ public void testGetJdbcUrl() {
+ Configuration conf = new Configuration();
+ conf.set(HConstants.ZOOKEEPER_QUORUM, "myzkclient:2181");
+ String jdbcUrl = CsvToKeyValueMapper.getJdbcUrl(conf);
+
+ assertEquals("jdbc:phoenix:myzkclient:2181", jdbcUrl);
+ }
+
+ @Test(expected=IllegalStateException.class)
+ public void testGetJdbcUrl_NotConfigured() {
+ Configuration conf = new Configuration();
+ CsvToKeyValueMapper.getJdbcUrl(conf);
+ }
+
+ @Test
+ public void testLoadPreUpdateProcessor() {
+ Configuration conf = new Configuration();
+ conf.setClass(CsvToKeyValueMapper.UPSERT_HOOK_CLASS_CONFKEY, MockUpsertProcessor.class,
+ ImportPreUpsertKeyValueProcessor.class);
+
+ ImportPreUpsertKeyValueProcessor processor = CsvToKeyValueMapper.loadPreUpsertProcessor(conf);
+ assertEquals(MockUpsertProcessor.class, processor.getClass());
+ }
+
+ @Test
+ public void testLoadPreUpdateProcessor_NotConfigured() {
+
+ Configuration conf = new Configuration();
+ ImportPreUpsertKeyValueProcessor processor = CsvToKeyValueMapper.loadPreUpsertProcessor(conf);
+
+ assertEquals(CsvToKeyValueMapper.DefaultImportPreUpsertKeyValueProcessor.class,
+ processor.getClass());
+ }
+
+ @Test(expected=IllegalStateException.class)
+ public void testLoadPreUpdateProcessor_ClassNotFound() {
+ Configuration conf = new Configuration();
+ conf.set(CsvToKeyValueMapper.UPSERT_HOOK_CLASS_CONFKEY, "MyUndefinedClass");
+
+ CsvToKeyValueMapper.loadPreUpsertProcessor(conf);
+ }
+
+
+ static class MockUpsertProcessor implements ImportPreUpsertKeyValueProcessor {
+
+ @Override
+ public List<KeyValue> preUpsert(byte[] rowKey, List<KeyValue> keyValues) {
+ throw new UnsupportedOperationException("Not yet implemented");
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/ed9f58ba/phoenix-core/src/test/java/org/apache/phoenix/util/ColumnInfoTest.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/test/java/org/apache/phoenix/util/ColumnInfoTest.java b/phoenix-core/src/test/java/org/apache/phoenix/util/ColumnInfoTest.java
new file mode 100644
index 0000000..85f3b44
--- /dev/null
+++ b/phoenix-core/src/test/java/org/apache/phoenix/util/ColumnInfoTest.java
@@ -0,0 +1,44 @@
+/*
+ * 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.
+ */
+package org.apache.phoenix.util;
+
+import org.apache.phoenix.schema.IllegalDataException;
+import org.junit.Test;
+
+import java.sql.Types;
+
+import static org.junit.Assert.assertEquals;
+
+public class ColumnInfoTest {
+
+ @Test
+ public void testToFromStringRoundTrip() {
+ ColumnInfo columnInfo = new ColumnInfo("myColumn", Types.INTEGER);
+ assertEquals(columnInfo, ColumnInfo.fromString(columnInfo.toString()));
+ }
+
+ @Test(expected=IllegalArgumentException.class)
+ public void testFromString_InvalidString() {
+ ColumnInfo.fromString("invalid");
+ }
+
+ @Test(expected= IllegalDataException.class)
+ public void testFromString_InvalidDataType() {
+ ColumnInfo.fromString("COLNAME:badType");
+ }
+}
[2/2] git commit: PHOENIX-129 Improve MapReduce-based import
(GabrielReid)
Posted by ja...@apache.org.
PHOENIX-129 Improve MapReduce-based import (GabrielReid)
Project: http://git-wip-us.apache.org/repos/asf/incubator-phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-phoenix/commit/ed9f58ba
Tree: http://git-wip-us.apache.org/repos/asf/incubator-phoenix/tree/ed9f58ba
Diff: http://git-wip-us.apache.org/repos/asf/incubator-phoenix/diff/ed9f58ba
Branch: refs/heads/3.0
Commit: ed9f58bae048ccc1e5653808e59d6c3231c91529
Parents: 6451b1d
Author: James Taylor <ja...@apache.org>
Authored: Sun Mar 16 11:21:53 2014 -0700
Committer: James Taylor <ja...@apache.org>
Committed: Sun Mar 16 11:21:53 2014 -0700
----------------------------------------------------------------------
bin/csv-bulk-loader.py | 52 ---
bin/readme.txt | 38 +-
phoenix-assembly/pom.xml | 22 +-
phoenix-assembly/src/build/all.xml | 3 +-
phoenix-assembly/src/build/mapreduce.xml | 44 ++
.../phoenix/mapreduce/CsvBulkLoadToolIT.java | 140 +++++++
.../phoenix/map/reduce/CSVBulkLoader.java | 399 -------------------
.../apache/phoenix/map/reduce/MapReduceJob.java | 202 ----------
.../phoenix/map/reduce/util/ConfigReader.java | 112 ------
.../phoenix/mapreduce/CsvBulkImportUtil.java | 67 ++++
.../phoenix/mapreduce/CsvBulkLoadTool.java | 341 ++++++++++++++++
.../phoenix/mapreduce/CsvToKeyValueMapper.java | 307 ++++++++++++++
.../ImportPreUpsertKeyValueProcessor.java | 47 +++
.../org/apache/phoenix/schema/PDataType.java | 2 +-
.../apache/phoenix/util/CSVCommonsLoader.java | 23 +-
.../org/apache/phoenix/util/ColumnInfo.java | 70 +++-
.../mapreduce/CsvBulkImportUtilTest.java | 71 ++++
.../phoenix/mapreduce/CsvBulkLoadToolTest.java | 80 ++++
.../mapreduce/CsvToKeyValueMapperTest.java | 132 ++++++
.../org/apache/phoenix/util/ColumnInfoTest.java | 44 ++
20 files changed, 1392 insertions(+), 804 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/ed9f58ba/bin/csv-bulk-loader.py
----------------------------------------------------------------------
diff --git a/bin/csv-bulk-loader.py b/bin/csv-bulk-loader.py
deleted file mode 100755
index 385ef41..0000000
--- a/bin/csv-bulk-loader.py
+++ /dev/null
@@ -1,52 +0,0 @@
-#!/usr/bin/env python
-############################################################################
-#
-# 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.
-#
-############################################################################
-
-# commandline-options
-# -i CSV data file path in hdfs (mandatory)
-# -s Phoenix schema name (mandatory if table is
-# created without default phoenix schema name)
-# -t Phoenix table name (mandatory)
-# -sql Phoenix create table ddl path (mandatory)
-# -zk Zookeeper IP:<port> (mandatory)
-# -hd HDFS NameNode IP:<port> (mandatory)
-# -mr MapReduce Job Tracker IP:<port> (mandatory)
-# -o Output directory path in hdfs (optional)
-# -idx Phoenix index table name (optional, index
-# support is yet to be added)
-# -error Ignore error while reading rows from CSV ?
-# (1 - YES | 0 - NO, defaults to 1) (optional)
-# -help Print all options (optional)
-
-import os
-import subprocess
-import sys
-import phoenix_utils
-
-current_dir = os.path.dirname(os.path.abspath(__file__))
-phoenix_jar_path = os.path.join(current_dir, "..", "phoenix-assembly",
- "target")
-phoenix_client_jar = phoenix_utils.find("phoenix-*-client.jar", phoenix_jar_path)
-
-java_cmd = 'java -cp ".' + os.pathsep + phoenix_client_jar +\
- '" org.apache.phoenix.map.reduce.CSVBulkLoader ' +\
- ' '.join(sys.argv[1:])
-
-subprocess.call(java_cmd, shell=True)
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/ed9f58ba/bin/readme.txt
----------------------------------------------------------------------
diff --git a/bin/readme.txt b/bin/readme.txt
index fa23eeb..2b57e8c 100644
--- a/bin/readme.txt
+++ b/bin/readme.txt
@@ -31,26 +31,20 @@ Usage: performance <zookeeper> <row count>
Example: Generates and upserts 1000000 rows and time basic queries on this data
./performance.sh localhost 1000000
-csv-bulk-loader.sh
-==================
-
-Usage: csv-bulk-loader <option value>
-Note: phoenix-[version].jar needs to be on Hadoop classpath on each node
-
-<option> <value>
--i CSV data file path in hdfs (mandatory)
--s Phoenix schema name (mandatory if not default)
--t Phoenix table name (mandatory)
--sql Phoenix create table sql file path (mandatory)
--zk Zookeeper IP:<port> (mandatory)
--mr MapReduce Job Tracker IP:<port> (mandatory)
--hd HDFS NameNode IP:<port> (mandatory)
--o Output directory path in hdfs (optional)
--idx Phoenix index table name (optional, not yet supported)
--error Ignore error while reading rows from CSV? (1-YES | 0-NO, default-1) (optional)
--help Print all options (optional)
-
-Example:
-./csv-bulk-loader.sh -i hdfs://server:9000/mydir/data.csv -s ns -t example -sql ~/Documents/createTable.sql -zk server:2181 -hd hdfs://server:9000 -mr server:9001
-
+CSV MapReduce Loader
+====================
+
+Usage: hadoop jar phoneix-[version]-mapreduce.jar <parameters>
+
+ -a,--array-delimiter <arg> Array element delimiter (optional)
+ -c,--import-columns <arg> Comma-separated list of columns to be
+ imported
+ -d,--delimiter <arg> Input delimiter, defaults to comma
+ -g,--ignore-errors Ignore input errors
+ -h,--help Show this help and quit
+ -i,--input <arg> Input CSV path (mandatory)
+ -o,--output <arg> Output path for temporary HFiles (optional)
+ -s,--schema <arg> Phoenix schema name (optional)
+ -t,--table <arg> Phoenix table name (mandatory)
+ -z,--zookeeper <arg> Zookeeper quorum to connect to (optional)
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/ed9f58ba/phoenix-assembly/pom.xml
----------------------------------------------------------------------
diff --git a/phoenix-assembly/pom.xml b/phoenix-assembly/pom.xml
index 9a69cab..d88ec76 100644
--- a/phoenix-assembly/pom.xml
+++ b/phoenix-assembly/pom.xml
@@ -62,6 +62,26 @@
</configuration>
</execution>
<execution>
+ <id>mapreduce-jar</id>
+ <phase>package</phase>
+ <goals>
+ <goal>single</goal>
+ </goals>
+ <configuration>
+ <finalName>phoenix-${project.version}</finalName>
+ <attach>false</attach>
+ <appendAssemblyId>true</appendAssemblyId>
+ <descriptors>
+ <descriptor>src/build/mapreduce.xml</descriptor>
+ </descriptors>
+ <archive>
+ <manifest>
+ <mainClass>org.apache.phoenix.mapreduce.CsvBulkLoadTool</mainClass>
+ </manifest>
+ </archive>
+ </configuration>
+ </execution>
+ <execution>
<id>package-to-tar</id>
<phase>package</phase>
<goals>
@@ -84,7 +104,7 @@
<goal>single</goal>
</goals>
<configuration>
- <finalName>phoenix-${project.version}</finalName>
+ <finalName>phoenix-${project.version}</finalName>
<attach>false</attach>
<appendAssemblyId>true</appendAssemblyId>
<descriptors>
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/ed9f58ba/phoenix-assembly/src/build/all.xml
----------------------------------------------------------------------
diff --git a/phoenix-assembly/src/build/all.xml b/phoenix-assembly/src/build/all.xml
index 9c1bc41..c4f7f61 100644
--- a/phoenix-assembly/src/build/all.xml
+++ b/phoenix-assembly/src/build/all.xml
@@ -110,13 +110,14 @@
<fileMode>0644</fileMode>
<directoryMode>0755</directoryMode>
</fileSet>
- <!-- Add the client jar. Expects the client jar packaging phase to already be run,
+ <!-- Add the client & mapreduce jars. Expects the client jar packaging phase to already be run,
which is determined by specification order in the pom. -->
<fileSet>
<directory>target</directory>
<outputDirectory>/</outputDirectory>
<includes>
<include>phoenix-*-client.jar</include>
+ <include>phoenix-*-mapreduce.jar</include>
</includes>
</fileSet>
</fileSets>
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/ed9f58ba/phoenix-assembly/src/build/mapreduce.xml
----------------------------------------------------------------------
diff --git a/phoenix-assembly/src/build/mapreduce.xml b/phoenix-assembly/src/build/mapreduce.xml
new file mode 100644
index 0000000..1e36748
--- /dev/null
+++ b/phoenix-assembly/src/build/mapreduce.xml
@@ -0,0 +1,44 @@
+<!--
+
+ 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.
+
+-->
+
+<assembly>
+ <id>mapreduce</id>
+ <formats>
+ <format>jar</format>
+ </formats>
+ <includeBaseDirectory>false</includeBaseDirectory>
+ <dependencySets>
+ <dependencySet>
+ <unpack>false</unpack>
+ <scope>runtime</scope>
+ <outputDirectory>lib</outputDirectory>
+ <excludes>
+ <exclude>${groupId}:phoenix-*</exclude>
+ </excludes>
+ </dependencySet>
+ <dependencySet>
+ <unpack>true</unpack>
+ <includes>
+ <include>${groupId}:phoenix-core</include>
+ </includes>
+ </dependencySet>
+ </dependencySets>
+</assembly>
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/ed9f58ba/phoenix-core/src/it/java/org/apache/phoenix/mapreduce/CsvBulkLoadToolIT.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/mapreduce/CsvBulkLoadToolIT.java b/phoenix-core/src/it/java/org/apache/phoenix/mapreduce/CsvBulkLoadToolIT.java
new file mode 100644
index 0000000..c0463bd
--- /dev/null
+++ b/phoenix-core/src/it/java/org/apache/phoenix/mapreduce/CsvBulkLoadToolIT.java
@@ -0,0 +1,140 @@
+/*
+ * 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.
+ */
+package org.apache.phoenix.mapreduce;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hbase.HBaseTestingUtility;
+import org.apache.phoenix.jdbc.PhoenixDriver;
+import org.apache.phoenix.util.PhoenixRuntime;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import java.io.PrintWriter;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.ResultSet;
+import java.sql.Statement;
+
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+public class CsvBulkLoadToolIT {
+
+ // We use HBaseTestUtil because we need to start up a MapReduce cluster as well
+ private static HBaseTestingUtility hbaseTestUtil;
+ private static String zkQuorum;
+ private static Connection conn;
+
+ @BeforeClass
+ public static void setUp() throws Exception {
+ hbaseTestUtil = new HBaseTestingUtility();
+ Configuration conf = hbaseTestUtil.getConfiguration();
+ hbaseTestUtil.startMiniCluster();
+ hbaseTestUtil.startMiniMapReduceCluster();
+
+ Class.forName(PhoenixDriver.class.getName());
+ zkQuorum = "localhost:" + hbaseTestUtil.getZkCluster().getClientPort();
+ conn = DriverManager.getConnection(PhoenixRuntime.JDBC_PROTOCOL
+ + PhoenixRuntime.JDBC_PROTOCOL_SEPARATOR + zkQuorum);
+ }
+
+ @AfterClass
+ public static void tearDownAfterClass() throws Exception {
+ conn.close();
+ PhoenixDriver.INSTANCE.close();
+ hbaseTestUtil.shutdownMiniCluster();
+ }
+
+ @Test
+ public void testBasicImport() throws Exception {
+
+ Statement stmt = conn.createStatement();
+ stmt.execute("CREATE TABLE TABLE1 (ID INTEGER NOT NULL PRIMARY KEY, NAME VARCHAR)");
+
+ FileSystem fs = FileSystem.get(hbaseTestUtil.getConfiguration());
+ FSDataOutputStream outputStream = fs.create(new Path("/tmp/input1.csv"));
+ PrintWriter printWriter = new PrintWriter(outputStream);
+ printWriter.println("1,Name 1");
+ printWriter.println("2,Name 2");
+ printWriter.close();
+
+ CsvBulkLoadTool csvBulkLoadTool = new CsvBulkLoadTool();
+ csvBulkLoadTool.setConf(hbaseTestUtil.getConfiguration());
+ int exitCode = csvBulkLoadTool.run(new String[] {
+ "--input", "/tmp/input1.csv",
+ "--table", "table1",
+ "--zookeeper", zkQuorum});
+ assertEquals(0, exitCode);
+
+ ResultSet rs = stmt.executeQuery("SELECT id, name FROM table1 ORDER BY id");
+ assertTrue(rs.next());
+ assertEquals(1, rs.getInt(1));
+ assertEquals("Name 1", rs.getString(2));
+ assertTrue(rs.next());
+ assertEquals(2, rs.getInt(1));
+ assertEquals("Name 2", rs.getString(2));
+ assertFalse(rs.next());
+
+ rs.close();
+ stmt.close();
+ }
+
+ @Test
+ public void testFullOptionImport() throws Exception {
+
+ Statement stmt = conn.createStatement();
+ stmt.execute("CREATE TABLE TABLE2 (ID INTEGER NOT NULL PRIMARY KEY, " +
+ "NAME VARCHAR, NAMES VARCHAR ARRAY)");
+
+ FileSystem fs = FileSystem.get(hbaseTestUtil.getConfiguration());
+ FSDataOutputStream outputStream = fs.create(new Path("/tmp/input2.csv"));
+ PrintWriter printWriter = new PrintWriter(outputStream);
+ printWriter.println("1|Name 1a;Name 1b");
+ printWriter.println("2|Name 2a;Name 2b");
+ printWriter.close();
+
+ CsvBulkLoadTool csvBulkLoadTool = new CsvBulkLoadTool();
+ csvBulkLoadTool.setConf(hbaseTestUtil.getConfiguration());
+ int exitCode = csvBulkLoadTool.run(new String[] {
+ "--input", "/tmp/input2.csv",
+ "--table", "table2",
+ "--zookeeper", zkQuorum,
+ "--delimiter", "|",
+ "--array-delimiter", ";",
+ "--import-columns", "ID,NAMES"});
+ assertEquals(0, exitCode);
+
+ ResultSet rs = stmt.executeQuery("SELECT id, names FROM table2 ORDER BY id");
+ assertTrue(rs.next());
+ assertEquals(1, rs.getInt(1));
+ assertArrayEquals(new Object[] { "Name 1a", "Name 1b" }, (Object[]) rs.getArray(2).getArray());
+ assertTrue(rs.next());
+ assertEquals(2, rs.getInt(1));
+ assertArrayEquals(new Object[] { "Name 2a", "Name 2b" }, (Object[]) rs.getArray(2).getArray());
+ assertFalse(rs.next());
+
+ rs.close();
+ stmt.close();
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/ed9f58ba/phoenix-core/src/main/java/org/apache/phoenix/map/reduce/CSVBulkLoader.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/map/reduce/CSVBulkLoader.java b/phoenix-core/src/main/java/org/apache/phoenix/map/reduce/CSVBulkLoader.java
deleted file mode 100644
index 0fd74e1..0000000
--- a/phoenix-core/src/main/java/org/apache/phoenix/map/reduce/CSVBulkLoader.java
+++ /dev/null
@@ -1,399 +0,0 @@
-/*
- * 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.
- */
-package org.apache.phoenix.map.reduce;
-
-import java.io.BufferedReader;
-import java.io.BufferedWriter;
-import java.io.FileReader;
-import java.io.FileWriter;
-import java.io.IOException;
-import java.sql.Connection;
-import java.sql.DriverManager;
-import java.sql.PreparedStatement;
-import java.sql.ResultSet;
-import java.sql.SQLException;
-import java.util.Date;
-import java.util.Map;
-
-import org.apache.commons.cli.CommandLine;
-import org.apache.commons.cli.CommandLineParser;
-import org.apache.commons.cli.HelpFormatter;
-import org.apache.commons.cli.Options;
-import org.apache.commons.cli.PosixParser;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.hbase.KeyValue;
-import org.apache.hadoop.hbase.client.HTable;
-import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
-import org.apache.hadoop.hbase.mapreduce.HFileOutputFormat;
-import org.apache.hadoop.hbase.mapreduce.LoadIncrementalHFiles;
-import org.apache.hadoop.hbase.regionserver.metrics.SchemaMetrics;
-import org.apache.hadoop.mapreduce.Job;
-import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
-import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
-import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
-
-import com.google.common.io.Closeables;
-import org.apache.phoenix.map.reduce.util.ConfigReader;
-import org.apache.phoenix.query.QueryConstants;
-import org.apache.phoenix.util.PhoenixRuntime;
-import org.apache.phoenix.util.SchemaUtil;
-import org.apache.phoenix.util.StringUtil;
-
-public class CSVBulkLoader {
- private static final String UNDERSCORE = "_";
-
- static FileWriter wr = null;
- static BufferedWriter bw = null;
- static boolean isDebug = false; //Set to true, if you need to log the bulk-import time.
- static ConfigReader systemConfig = null;
-
- static String schemaName = "";
- static String tableName = "";
- static String idxTable = "";
- static String createPSQL[] = null;
- static String skipErrors = null;
- static String zookeeperIP = null;
- static String mapredIP = null;
- static String hdfsNameNode = null;
-
- static{
- /** load the log-file writer, if debug is true **/
- if(isDebug){
- try {
- wr = new FileWriter("phoenix-bulk-import.log", false);
- bw = new BufferedWriter(wr);
- } catch (IOException e) {
- System.err.println("Error preparing writer for log file :: " + e.getMessage());
- }
- }
-
- /** load the Map-Reduce configs **/
- try {
- systemConfig = new ConfigReader("csv-bulk-load-config.properties");
- } catch (Exception e) {
- System.err.println("Exception occurred while reading config properties");
- System.err.println("The bulk loader will run slower than estimated");
- }
- }
-
- /**
- * -i CSV data file path in hdfs
- * -s Phoenix schema name
- * -t Phoenix table name
- * -sql Phoenix create table sql path (1 SQL statement per line)
- * -zk Zookeeper IP:<port>
- * -mr MapReduce Job Tracker IP:<port>
- * -hd HDFS NameNode IP:<port>
- * -o Output directory path in hdfs (Optional)
- * -idx Phoenix index table name (Optional)
- * -error Ignore error while reading rows from CSV ? (1 - YES/0 - NO, defaults to 1) (OPtional)
- * -help Print all options (Optional)
- */
-
- @SuppressWarnings("deprecation")
- public static void main(String[] args) throws Exception{
-
- String inputFile = null;
- String outFile = null;
-
- Options options = new Options();
- options.addOption("i", true, "CSV data file path");
- options.addOption("o", true, "Output directory path");
- options.addOption("s", true, "Phoenix schema name");
- options.addOption("t", true, "Phoenix table name");
- options.addOption("idx", true, "Phoenix index table name");
- options.addOption("zk", true, "Zookeeper IP:<port>");
- options.addOption("mr", true, "MapReduce Job Tracker IP:<port>");
- options.addOption("hd", true, "HDFS NameNode IP:<port>");
- options.addOption("sql", true, "Phoenix create table sql path");
- options.addOption("error", true, "Ignore error while reading rows from CSV ? (1 - YES/0 - NO, defaults to 1)");
- options.addOption("help", false, "All options");
-
- CommandLineParser parser = new PosixParser();
- CommandLine cmd = parser.parse( options, args);
-
- if(cmd.hasOption("help")){
- HelpFormatter formatter = new HelpFormatter();
- formatter.printHelp( "help", options );
- System.exit(0);
- }
-
- String parser_error = "ERROR while parsing arguments. ";
- //CSV input, table name, sql and zookeeper IP are mandatory fields
- if(cmd.hasOption("i")){
- inputFile = cmd.getOptionValue("i");
- }else{
- System.err.println(parser_error + "Please provide CSV file input path");
- System.exit(0);
- }
- if(cmd.hasOption("t")){
- tableName = cmd.getOptionValue("t");
- }else{
- System.err.println(parser_error + "Please provide Phoenix table name");
- System.exit(0);
- }
- if(cmd.hasOption("sql")){
- String sqlPath = cmd.getOptionValue("sql");
- createPSQL = getCreatePSQLstmts(sqlPath);
- }
- if(cmd.hasOption("zk")){
- zookeeperIP = cmd.getOptionValue("zk");
- }else{
- System.err.println(parser_error + "Please provide Zookeeper address");
- System.exit(0);
- }
- if(cmd.hasOption("mr")){
- mapredIP = cmd.getOptionValue("mr");
- }else{
- System.err.println(parser_error + "Please provide MapReduce address");
- System.exit(0);
- }
- if(cmd.hasOption("hd")){
- hdfsNameNode = cmd.getOptionValue("hd");
- }else{
- System.err.println(parser_error + "Please provide HDFS NameNode address");
- System.exit(0);
- }
-
- if(cmd.hasOption("o")){
- outFile = cmd.getOptionValue("o");
- }else{
- outFile = "phoenix-output-dir";
- }
- if(cmd.hasOption("s")){
- schemaName = cmd.getOptionValue("s");
- }
- if(cmd.hasOption("idx")){
- idxTable = cmd.getOptionValue("idx");
- }
- if(cmd.hasOption("error")){
- skipErrors = cmd.getOptionValue("error");
- }else{
- skipErrors = "1";
- }
-
- log("[TS - START] :: " + new Date() + "\n");
-
- Path inputPath = new Path(inputFile);
- Path outPath = new Path(outFile);
-
- //Create the Phoenix table in HBase
- if (createPSQL != null) {
- for(String s : createPSQL){
- if(s == null || s.trim().length() == 0) {
- continue;
- }
- createTable(s);
- }
-
- log("[TS - Table created] :: " + new Date() + "\n");
- }
-
- String dataTable = "";
- if(schemaName != null && schemaName.trim().length() > 0)
- dataTable = SchemaUtil.normalizeIdentifier(schemaName) + "." + SchemaUtil.normalizeIdentifier(tableName);
- else
- dataTable = SchemaUtil.normalizeIdentifier(tableName);
-
- try {
- validateTable();
- } catch (SQLException e) {
- System.err.println(e.getMessage());
- System.exit(0);
- }
-
- Configuration conf = new Configuration();
- loadMapRedConfigs(conf);
-
- Job job = new Job(conf, "MapReduce - Phoenix bulk import");
- job.setJarByClass(MapReduceJob.class);
- job.setInputFormatClass(TextInputFormat.class);
- FileInputFormat.addInputPath(job, inputPath);
-
- FileSystem fs = FileSystem.get(conf);
- fs.delete(outPath);
- FileOutputFormat.setOutputPath(job, outPath);
-
- job.setMapperClass(MapReduceJob.PhoenixMapper.class);
- job.setMapOutputKeyClass(ImmutableBytesWritable.class);
- job.setMapOutputValueClass(KeyValue.class);
-
- SchemaMetrics.configureGlobally(conf);
-
- HTable hDataTable = new HTable(conf, dataTable);
-
- // Auto configure partitioner and reducer according to the Main Data table
- HFileOutputFormat.configureIncrementalLoad(job, hDataTable);
-
- job.waitForCompletion(true);
-
- log("[TS - M-R HFile generated..Now dumping to HBase] :: " + new Date() + "\n");
-
- LoadIncrementalHFiles loader = new LoadIncrementalHFiles(conf);
- loader.doBulkLoad(new Path(outFile), hDataTable);
-
- log("[TS - FINISH] :: " + new Date() + "\n");
- if(isDebug) bw.close();
-
- }
-
- private static void createTable(String stmt) {
-
- Connection conn = null;
- PreparedStatement statement = null;
-
- try {
- conn = DriverManager.getConnection(getUrl(), "", "");
- try {
- statement = conn.prepareStatement(stmt);
- statement.execute();
- conn.commit();
- } finally {
- if(statement != null) {
- statement.close();
- }
- }
- } catch (Exception e) {
- System.err.println("Error creating the table :: " + e.getMessage());
- } finally{
- try {
- if(conn != null) {
- conn.close();
- }
- } catch (Exception e) {
- System.err.println("Failed to close connection :: " + e.getMessage());
- }
- }
- }
-
- /**
- * Perform any required validation on the table being bulk loaded into:
- * - ensure no column family names start with '_', as they'd be ignored leading to problems.
- * @throws SQLException
- */
- private static void validateTable() throws SQLException {
-
- Connection conn = DriverManager.getConnection(getUrl());
- try {
- ResultSet rs = conn.getMetaData().getColumns(null, StringUtil.escapeLike(schemaName), StringUtil.escapeLike(tableName), null);
- while (rs.next()) {
- String familyName = rs.getString(1);
- if (familyName != null && familyName.startsWith(UNDERSCORE)) {
- String msg;
- if (QueryConstants.DEFAULT_COLUMN_FAMILY.equals(familyName)) {
- msg = "CSV Bulk Loader error: All column names that are not part of the primary key constraint must be prefixed with a column family name (i.e. f.my_column VARCHAR)";
- } else {
- msg = "CSV Bulk Loader error: Column family name must not start with '_': " + familyName;
- }
- throw new SQLException(msg);
- }
- }
- } finally{
- conn.close();
- }
- }
-
- private static String getUrl() {
- return PhoenixRuntime.JDBC_PROTOCOL + PhoenixRuntime.JDBC_PROTOCOL_SEPARATOR + zookeeperIP;
- }
-
- private static void loadMapRedConfigs(Configuration conf){
-
- conf.set("IGNORE.INVALID.ROW", skipErrors);
- conf.set("schemaName", schemaName);
- conf.set("tableName", tableName);
- conf.set("zk", zookeeperIP);
- conf.set("hbase.zookeeper.quorum", zookeeperIP);
- conf.set("fs.default.name", hdfsNameNode);
- conf.set("mapred.job.tracker", mapredIP);
-
- //Load the other System-Configs
- try {
-
- Map<String, String> configs = systemConfig.getAllConfigMap();
-
- if(configs.containsKey("mapreduce.map.output.compress")){
- String s = configs.get("mapreduce.map.output.compress");
- if(s != null && s.trim().length() > 0)
- conf.set("mapreduce.map.output.compress", s);
- }
-
- if(configs.containsKey("mapreduce.map.output.compress.codec")){
- String s = configs.get("mapreduce.map.output.compress.codec");
- if(s != null && s.trim().length() > 0)
- conf.set("mapreduce.map.output.compress.codec", s);
- }
-
- if(configs.containsKey("io.sort.record.percent")){
- String s = configs.get("io.sort.record.percent");
- if(s != null && s.trim().length() > 0)
- conf.set("io.sort.record.percent", s);
- }
-
- if(configs.containsKey("io.sort.factor")){
- String s = configs.get("io.sort.factor");
- if(s != null && s.trim().length() > 0)
- conf.set("io.sort.factor", s);
- }
-
- if(configs.containsKey("mapred.tasktracker.map.tasks.maximum")){
- String s = configs.get("mapred.tasktracker.map.tasks.maximum");
- if(s != null && s.trim().length() > 0)
- conf.set("mapred.tasktracker.map.tasks.maximum", s);
- }
-
- } catch (Exception e) {
- System.err.println("Error loading the configs :: " + e.getMessage());
- System.err.println("The bulk loader will run slower than estimated");
- }
- }
-
- private static String[] getCreatePSQLstmts(String path){
-
- BufferedReader br = null;
- try {
- FileReader file = new FileReader(path);
- br = new BufferedReader(file);
- //Currently, we can have at-most 2 SQL statements - 1 for create table and 1 for index
- String[] sb = new String[2];
- String line;
- for(int i = 0; i < 2 && (line = br.readLine()) != null ; i++){
- sb[i] = line;
- }
- return sb;
-
- } catch (IOException e) {
- System.err.println("Error reading the file :: " + path + ", " + e.getMessage());
- } finally {
- if (br != null) Closeables.closeQuietly(br);
- }
- return null;
- }
-
- private static void log(String msg){
- if(isDebug){
- try {
- bw.write(msg);
- } catch (IOException e) {
- System.err.println("Error logging the statement :: " + msg);
- }
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/ed9f58ba/phoenix-core/src/main/java/org/apache/phoenix/map/reduce/MapReduceJob.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/map/reduce/MapReduceJob.java b/phoenix-core/src/main/java/org/apache/phoenix/map/reduce/MapReduceJob.java
deleted file mode 100644
index 9dc8032..0000000
--- a/phoenix-core/src/main/java/org/apache/phoenix/map/reduce/MapReduceJob.java
+++ /dev/null
@@ -1,202 +0,0 @@
-/*
- * 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.
- */
-package org.apache.phoenix.map.reduce;
-
-import java.io.ByteArrayInputStream;
-import java.io.IOException;
-import java.io.InputStreamReader;
-import java.sql.Connection;
-import java.sql.DriverManager;
-import java.sql.PreparedStatement;
-import java.sql.ResultSet;
-import java.sql.SQLException;
-import java.util.ArrayList;
-import java.util.Iterator;
-import java.util.LinkedHashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Properties;
-
-import org.apache.commons.csv.CSVFormat;
-import org.apache.commons.csv.CSVParser;
-import org.apache.commons.csv.CSVRecord;
-import org.apache.commons.lang.StringUtils;
-import org.apache.hadoop.hbase.KeyValue;
-import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
-import org.apache.hadoop.hbase.util.Pair;
-import org.apache.hadoop.io.LongWritable;
-import org.apache.hadoop.io.Text;
-import org.apache.hadoop.mapreduce.Mapper;
-import org.apache.phoenix.schema.PDataType;
-import org.apache.phoenix.util.PhoenixRuntime;
-import org.apache.phoenix.util.QueryUtil;
-
-public class MapReduceJob {
-
- public static class PhoenixMapper extends Mapper<LongWritable, Text, ImmutableBytesWritable, KeyValue>{
-
- private Connection conn_zk = null;
- private PreparedStatement[] stmtCache;
- private String tableName;
- private String schemaName;
- Map<Integer, Integer> colDetails = new LinkedHashMap<Integer, Integer>();
- boolean ignoreUpsertError = true;
- private String zookeeperIP;
-
- /**
- * Get the phoenix jdbc connection.
- */
-
- private static String getUrl(String url) {
- return PhoenixRuntime.JDBC_PROTOCOL + PhoenixRuntime.JDBC_PROTOCOL_SEPARATOR + url;
- }
-
- /***
- * Get the column information from the table metaData.
- * Cretae a map of col-index and col-data-type.
- * Create the upsert Prepared Statement based on the map-size.
- */
-
- @Override
- public void setup(Context context) throws InterruptedException{
- Properties props = new Properties();
-
- try {
- zookeeperIP = context.getConfiguration().get("zk");
-
- //ZK connection used to get the table meta-data
- conn_zk = DriverManager.getConnection(getUrl(zookeeperIP), props);
-
- schemaName = context.getConfiguration().get("schemaName");
- tableName = context.getConfiguration().get("tableName");
- ignoreUpsertError = context.getConfiguration().get("IGNORE.INVALID.ROW").equalsIgnoreCase("0") ? false : true;
-
- //Get the resultset from the actual zookeeper connection. Connectionless mode throws "UnSupportedOperation" exception for this
- ResultSet rs = conn_zk.getMetaData().getColumns(null, schemaName, tableName, null);
- //This map holds the key-value pair of col-position and its data type
- int i = 1;
- while(rs.next()){
- colDetails.put(i, rs.getInt(QueryUtil.DATA_TYPE_POSITION));
- i++;
- }
-
- stmtCache = new PreparedStatement[colDetails.size()];
- ArrayList<String> cols = new ArrayList<String>();
- for(i = 0 ; i < colDetails.size() ; i++){
- cols.add("?");
- String prepValues = StringUtils.join(cols, ",");
- String upsertStmt = "";
- if(schemaName != null && schemaName.trim().length() > 0)
- upsertStmt = "upsert into " + schemaName + "." + tableName + " values (" + prepValues + ")";
- else
- upsertStmt = "upsert into " + tableName + " values (" + prepValues + ")";
- try {
- stmtCache[i] = conn_zk.prepareStatement(upsertStmt);
- } catch (SQLException e) {
- System.err.println("Error preparing the upsert statement" + e.getMessage());
- if(!ignoreUpsertError){
- throw (new InterruptedException(e.getMessage()));
- }
- }
- }
- } catch (SQLException e) {
- System.err.println("Error occurred in connecting to Phoenix HBase" + e.getMessage());
- }
-
- }
-
- /* Tokenize the text input line based on the "," delimeter.
- * TypeCast the token based on the col-data-type using the convertTypeSpecificValue API below.
- * Upsert the data. DO NOT COMMIT.
- * Use Phoenix's getUncommittedDataIterator API to parse the uncommited data to KeyValue pairs.
- * Emit the row-key and KeyValue pairs from Mapper to allow sorting based on row-key.
- * Finally, do connection.rollback( to preserve table state).
- */
-
- @Override
- public void map(LongWritable key, Text line, Context context) throws IOException, InterruptedException{
-
- CSVParser parser = new CSVParser(new InputStreamReader(new ByteArrayInputStream(line.toString().getBytes())),CSVFormat.EXCEL);
- try {
- CSVRecord record = parser.iterator().next();
-
- PreparedStatement upsertStatement;
- if(record.size() >= stmtCache.length){
- //If CVS values are more than the number of cols in the table, apply the col count cap
- upsertStatement = stmtCache[stmtCache.length - 1];
- }else{
- //Else, take the corresponding upsertStmt from cached array
- upsertStatement = stmtCache[record.size() - 1];
- }
-
- for(int i = 0 ; i < record.size() && i < colDetails.size() ;i++){
- upsertStatement.setObject(i+1, convertTypeSpecificValue(record.get(i), colDetails.get(new Integer(i+1))));
- }
-
- upsertStatement.execute();
- } catch (SQLException e) {
- System.err.println("Failed to upsert data in the Phoenix :: " + e.getMessage());
- if(!ignoreUpsertError){
- throw (new InterruptedException(e.getMessage()));
- }
- } catch (Exception e) {
- System.err.println("Failed to upsert data in the Phoenix :: " + e.getMessage());
- }finally {
- parser.close();
- }
-
- Iterator<Pair<byte[],List<KeyValue>>> dataIterator = null;
- try {
- dataIterator = PhoenixRuntime.getUncommittedDataIterator(conn_zk);
- } catch (SQLException e) {
- System.err.println("Failed to retrieve the data iterator for Phoenix table :: " + e.getMessage());
- }
-
- while(dataIterator != null && dataIterator.hasNext()){
- Pair<byte[],List<KeyValue>> row = dataIterator.next();
- for(KeyValue kv : row.getSecond()){
- context.write(new ImmutableBytesWritable(kv.getRow()), kv);
- }
- }
-
- try {
- conn_zk.rollback();
- } catch (SQLException e) {
- System.err.println("Transaction rollback failed.");
- }
- }
-
- /*
- * Do connection.close()
- */
-
- @Override
- public void cleanup(Context context) {
- try {
- conn_zk.close();
- } catch (SQLException e) {
- System.err.println("Failed to close the JDBC connection");
- }
- }
-
- private Object convertTypeSpecificValue(String s, Integer sqlType) throws Exception {
- return PDataType.fromTypeId(sqlType).toObject(s);
- }
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/ed9f58ba/phoenix-core/src/main/java/org/apache/phoenix/map/reduce/util/ConfigReader.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/map/reduce/util/ConfigReader.java b/phoenix-core/src/main/java/org/apache/phoenix/map/reduce/util/ConfigReader.java
deleted file mode 100644
index 1c94d6f..0000000
--- a/phoenix-core/src/main/java/org/apache/phoenix/map/reduce/util/ConfigReader.java
+++ /dev/null
@@ -1,112 +0,0 @@
-/*
- * 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.
- */
-package org.apache.phoenix.map.reduce.util;
-
-import java.util.Enumeration;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.ResourceBundle;
-
-/**
- * Class to read configs.
- *
- */
-
-public class ConfigReader
- {
-
- private String propertyFile = null;
- private boolean loaded = false;
- private static final Object _synObj = new Object();
- private Map<String, String> properties = new HashMap<String, String>();
- private Exception loadException = null;
-
- /**
- * Retrieves singleton config objects from a hashmap of stored objects,
- * creates these objects if they aren't in the hashmap.
- */
-
- public ConfigReader(String propertyFile) {
- this.propertyFile = propertyFile;
- }
-
- public void load() throws Exception {
- if (loaded) {
- if (loadException != null) {
- throw new Exception(loadException);
- }
- return;
- }
- synchronized (_synObj) {
- if (!loaded) {
- try {
- String tmpFile = propertyFile.trim();
- if (tmpFile.endsWith(".properties")) {
- tmpFile = tmpFile
- .substring(0, tmpFile.lastIndexOf("."));
- }
- ResourceBundle resource = ResourceBundle.getBundle(tmpFile);
- Enumeration<String> enm = resource.getKeys();
-
- while (enm.hasMoreElements()) {
- String key = enm.nextElement();
- String value = resource.getString(key);
- properties.put(key, value);
- }
- } catch (Exception e) {
- System.err
- .println("Exception while loading the config.properties file :: "
- + e.getMessage());
- loadException = e;
- loaded = true;
- throw e;
- }
- loaded = true;
- }
- }
- }
-
- public void addConfig(String key, String value) {
- try {
- load();
- } catch (Exception e) {
- System.err.println("ERROR :: " + e.getMessage());
- }
- properties.put(key, value);
- }
-
- public boolean hasConfig(String key) {
- try {
- load();
- } catch (Exception e) {
- System.err.println("ERROR :: " + e.getMessage());
- }
- return properties.containsKey(key);
- }
-
- public String getConfig(String key) throws Exception {
- load();
- return properties.get(key);
- }
-
- public Map<String, String> getAllConfigMap() throws Exception {
- load();
- return properties;
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/ed9f58ba/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/CsvBulkImportUtil.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/CsvBulkImportUtil.java b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/CsvBulkImportUtil.java
new file mode 100644
index 0000000..c27e95f
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/CsvBulkImportUtil.java
@@ -0,0 +1,67 @@
+/*
+ * 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.
+ */
+package org.apache.phoenix.mapreduce;
+
+import com.google.common.base.Preconditions;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.phoenix.util.ColumnInfo;
+
+import java.util.List;
+
+/**
+ * Collection of utility methods for setting up bulk import jobs.
+ */
+public class CsvBulkImportUtil {
+
+ /**
+ * Configure a job configuration for a bulk CSV import.
+ *
+ * @param conf job configuration to be set up
+ * @param tableName name of the table to be imported to, can include a schema name
+ * @param fieldDelimiter field delimiter character for the CSV input
+ * @param arrayDelimiter array delimiter character, can be null
+ * @param columnInfoList list of columns to be imported
+ * @param ignoreInvalidRows flag to ignore invalid input rows
+ */
+ public static void initCsvImportJob(Configuration conf, String tableName, char fieldDelimiter,
+ String arrayDelimiter, List<ColumnInfo> columnInfoList, boolean ignoreInvalidRows) {
+
+ Preconditions.checkNotNull(tableName);
+ Preconditions.checkNotNull(columnInfoList);
+ Preconditions.checkArgument(!columnInfoList.isEmpty(), "Column info list is empty");
+ conf.set(CsvToKeyValueMapper.TABLE_NAME_CONFKEY, tableName);
+ conf.set(CsvToKeyValueMapper.FIELD_DELIMITER_CONFKEY, String.valueOf(fieldDelimiter));
+ if (arrayDelimiter != null) {
+ conf.set(CsvToKeyValueMapper.ARRAY_DELIMITER_CONFKEY, arrayDelimiter);
+ }
+ CsvToKeyValueMapper.configureColumnInfoList(conf, columnInfoList);
+ conf.setBoolean(CsvToKeyValueMapper.IGNORE_INVALID_ROW_CONFKEY, ignoreInvalidRows);
+ }
+
+ /**
+ * Configure an {@link ImportPreUpsertKeyValueProcessor} for a CSV bulk import job.
+ *
+ * @param conf job configuration
+ * @param processorClass class to be used for performing pre-upsert processing
+ */
+ public static void configurePreUpsertProcessor(Configuration conf,
+ Class<? extends ImportPreUpsertKeyValueProcessor> processorClass) {
+ conf.setClass(CsvToKeyValueMapper.UPSERT_HOOK_CLASS_CONFKEY, processorClass,
+ ImportPreUpsertKeyValueProcessor.class);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/ed9f58ba/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/CsvBulkLoadTool.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/CsvBulkLoadTool.java b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/CsvBulkLoadTool.java
new file mode 100644
index 0000000..93e4dab
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/CsvBulkLoadTool.java
@@ -0,0 +1,341 @@
+/*
+ * 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.
+ */
+package org.apache.phoenix.mapreduce;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Splitter;
+import com.google.common.collect.Lists;
+import org.apache.commons.cli.CommandLine;
+import org.apache.commons.cli.CommandLineParser;
+import org.apache.commons.cli.HelpFormatter;
+import org.apache.commons.cli.Option;
+import org.apache.commons.cli.Options;
+import org.apache.commons.cli.ParseException;
+import org.apache.commons.cli.PosixParser;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.conf.Configured;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hbase.HConstants;
+import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.client.HTable;
+import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
+import org.apache.hadoop.hbase.mapreduce.HFileOutputFormat;
+import org.apache.hadoop.hbase.mapreduce.LoadIncrementalHFiles;
+import org.apache.hadoop.mapreduce.Job;
+import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
+import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
+import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
+import org.apache.hadoop.util.Tool;
+import org.apache.hadoop.util.ToolRunner;
+import org.apache.phoenix.jdbc.PhoenixDatabaseMetaData;
+import org.apache.phoenix.query.QueryConstants;
+import org.apache.phoenix.util.CSVCommonsLoader;
+import org.apache.phoenix.util.ColumnInfo;
+import org.apache.phoenix.util.PhoenixRuntime;
+import org.apache.phoenix.util.SchemaUtil;
+import org.apache.phoenix.util.StringUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.List;
+import java.util.UUID;
+
+/**
+ * Base tool for running MapReduce-based ingests of data.
+ */
+public class CsvBulkLoadTool extends Configured implements Tool {
+
+ private static final Logger LOG = LoggerFactory.getLogger(CsvBulkLoadTool.class);
+
+ static final Option ZK_QUORUM_OPT = new Option("z", "zookeeper", true, "Zookeeper quorum to connect to (optional)");
+ static final Option INPUT_PATH_OPT = new Option("i", "input", true, "Input CSV path (mandatory)");
+ static final Option OUTPUT_PATH_OPT = new Option("o", "output", true, "Output path for temporary HFiles (optional)");
+ static final Option SCHEMA_NAME_OPT = new Option("s", "schema", true, "Phoenix schema name (optional)");
+ static final Option TABLE_NAME_OPT = new Option("t", "table", true, "Phoenix table name (mandatory)");
+ static final Option DELIMITER_OPT = new Option("d", "delimiter", true, "Input delimiter, defaults to comma");
+ static final Option ARRAY_DELIMITER_OPT = new Option("a", "array-delimiter", true, "Array element delimiter (optional)");
+ static final Option IMPORT_COLUMNS_OPT = new Option("c", "import-columns", true, "Comma-separated list of columns to be imported");
+ static final Option IGNORE_ERRORS_OPT = new Option("g", "ignore-errors", false, "Ignore input errors");
+ static final Option HELP_OPT = new Option("h", "help", false, "Show this help and quit");
+
+ public static void main(String[] args) throws Exception {
+ ToolRunner.run(new CsvBulkLoadTool(), args);
+ }
+
+ /**
+ * Parses the commandline arguments, throws IllegalStateException if mandatory argumentsn are
+ * missing.
+ *
+ * @param args supplied command line arguments
+ * @return the parsed command line
+ */
+ CommandLine parseOptions(String[] args) {
+
+ Options options = getOptions();
+
+ CommandLineParser parser = new PosixParser();
+ CommandLine cmdLine = null;
+ try {
+ cmdLine = parser.parse(options, args);
+ } catch (ParseException e) {
+ printHelpAndExit("Error parsing command line options: " + e.getMessage(), options);
+ }
+
+ if (cmdLine.hasOption(HELP_OPT.getOpt())) {
+ printHelpAndExit(options);
+ }
+
+ if (!cmdLine.hasOption(TABLE_NAME_OPT.getOpt())) {
+ throw new IllegalStateException(TABLE_NAME_OPT.getLongOpt() + " is a mandatory " +
+ "parameter");
+ }
+
+ if (!cmdLine.getArgList().isEmpty()) {
+ throw new IllegalStateException("Got unexpected extra parameters: "
+ + cmdLine.getArgList());
+ }
+
+ if (!cmdLine.hasOption(INPUT_PATH_OPT.getOpt())) {
+ throw new IllegalStateException(INPUT_PATH_OPT.getLongOpt() + " is a mandatory " +
+ "parameter");
+ }
+
+ return cmdLine;
+ }
+
+ private Options getOptions() {
+ Options options = new Options();
+ options.addOption(INPUT_PATH_OPT);
+ options.addOption(TABLE_NAME_OPT);
+ options.addOption(ZK_QUORUM_OPT);
+ options.addOption(OUTPUT_PATH_OPT);
+ options.addOption(SCHEMA_NAME_OPT);
+ options.addOption(DELIMITER_OPT);
+ options.addOption(ARRAY_DELIMITER_OPT);
+ options.addOption(IMPORT_COLUMNS_OPT);
+ options.addOption(IGNORE_ERRORS_OPT);
+ options.addOption(HELP_OPT);
+ return options;
+ }
+
+
+ private void printHelpAndExit(String errorMessage, Options options) {
+ System.err.println(errorMessage);
+ printHelpAndExit(options);
+ }
+
+ private void printHelpAndExit(Options options) {
+ HelpFormatter formatter = new HelpFormatter();
+ formatter.printHelp("help", options);
+ System.exit(0);
+ }
+
+ @Override
+ public int run(String[] args) throws Exception {
+
+ CommandLine cmdLine = null;
+ try {
+ cmdLine = parseOptions(args);
+ } catch (IllegalStateException e) {
+ printHelpAndExit(e.getMessage(), getOptions());
+ }
+ Class.forName(DriverManager.class.getName());
+ Connection conn = DriverManager.getConnection(
+ getJdbcUrl(cmdLine.getOptionValue(ZK_QUORUM_OPT.getOpt())));
+ String tableName = cmdLine.getOptionValue(TABLE_NAME_OPT.getOpt());
+ String schemaName = cmdLine.getOptionValue(SCHEMA_NAME_OPT.getOpt());
+ String qualifiedTableName = getQualifiedTableName(schemaName, tableName);
+ List<ColumnInfo> importColumns = buildImportColumns(conn, cmdLine, qualifiedTableName);
+ configureOptions(cmdLine, importColumns, getConf());
+
+ try {
+ validateTable(conn,schemaName, tableName);
+ } finally {
+ conn.close();
+ }
+
+ Path inputPath = new Path(cmdLine.getOptionValue(INPUT_PATH_OPT.getOpt()));
+ Path outputPath = null;
+ if (cmdLine.hasOption(OUTPUT_PATH_OPT.getOpt())) {
+ outputPath = new Path(cmdLine.getOptionValue(OUTPUT_PATH_OPT.getOpt()));
+ } else {
+ outputPath = new Path("/tmp/" + UUID.randomUUID());
+ }
+ LOG.info("Configuring HFile output path to {}", outputPath);
+
+ Job job = new Job(getConf(),
+ "Phoenix MapReduce import for "
+ + getConf().get(CsvToKeyValueMapper.TABLE_NAME_CONFKEY));
+
+ // Allow overriding the job jar setting by using a -D system property at startup
+ if (job.getJar() == null) {
+ job.setJarByClass(CsvToKeyValueMapper.class);
+ }
+ job.setInputFormatClass(TextInputFormat.class);
+ FileInputFormat.addInputPath(job, inputPath);
+
+ FileSystem fs = FileSystem.get(getConf());
+ FileOutputFormat.setOutputPath(job, outputPath);
+
+ job.setMapperClass(CsvToKeyValueMapper.class);
+ job.setMapOutputKeyClass(ImmutableBytesWritable.class);
+ job.setMapOutputValueClass(KeyValue.class);
+
+ HTable htable = new HTable(getConf(), qualifiedTableName);
+
+ // Auto configure partitioner and reducer according to the Main Data table
+ HFileOutputFormat.configureIncrementalLoad(job, htable);
+
+ LOG.info("Running MapReduce import job from {} to {}", inputPath, outputPath);
+ boolean success = job.waitForCompletion(true);
+ if (!success) {
+ LOG.error("Import job failed, check JobTracker for details");
+ return 1;
+ }
+
+ LOG.info("Loading HFiles from {}", outputPath);
+ LoadIncrementalHFiles loader = new LoadIncrementalHFiles(getConf());
+ loader.doBulkLoad(outputPath, htable);
+ htable.close();
+
+ LOG.info("Incremental load complete");
+
+ LOG.info("Removing output directory {}", outputPath);
+ if (!FileSystem.get(getConf()).delete(outputPath, true)) {
+ LOG.error("Removing output directory {} failed", outputPath);
+ }
+
+ return 0;
+ }
+
+ String getJdbcUrl(String zkQuorum) {
+ if (zkQuorum == null) {
+ LOG.warn("Defaulting to localhost for ZooKeeper quorum");
+ zkQuorum = "localhost:2181";
+ }
+ return PhoenixRuntime.JDBC_PROTOCOL + PhoenixRuntime.JDBC_PROTOCOL_SEPARATOR + zkQuorum;
+ }
+
+ /**
+ * Build up the list of columns to be imported. The list is taken from the command line if
+ * present, otherwise it is taken from the table description.
+ *
+ * @param conn connection to Phoenix
+ * @param cmdLine supplied command line options
+ * @param qualifiedTableName table name (possibly with schema) of the table to be imported
+ * @return the list of columns to be imported
+ */
+ List<ColumnInfo> buildImportColumns(Connection conn, CommandLine cmdLine,
+ String qualifiedTableName) throws SQLException {
+ List<String> userSuppliedColumnNames = null;
+ if (cmdLine.hasOption(IMPORT_COLUMNS_OPT.getOpt())) {
+ userSuppliedColumnNames = Lists.newArrayList(
+ Splitter.on(",").trimResults().split
+ (cmdLine.getOptionValue(IMPORT_COLUMNS_OPT.getOpt())));
+ }
+ return CSVCommonsLoader.generateColumnInfo(
+ conn, qualifiedTableName, userSuppliedColumnNames, true);
+ }
+
+ /**
+ * Calculate the HBase HTable name for which the import is to be done.
+ *
+ * @param schemaName import schema name, can be null
+ * @param tableName import table name
+ * @return the byte representation of the import HTable
+ */
+ @VisibleForTesting
+ static String getQualifiedTableName(String schemaName, String tableName) {
+ if (schemaName != null) {
+ return String.format("%s.%s", SchemaUtil.normalizeIdentifier(schemaName),
+ SchemaUtil.normalizeIdentifier(tableName));
+ } else {
+ return SchemaUtil.normalizeIdentifier(tableName);
+ }
+ }
+
+ /**
+ * Set configuration values based on parsed command line options.
+ *
+ * @param cmdLine supplied command line options
+ * @param importColumns descriptors of columns to be imported
+ * @param conf job configuration
+ */
+ @VisibleForTesting
+ static void configureOptions(CommandLine cmdLine, List<ColumnInfo> importColumns,
+ Configuration conf) {
+
+ char delimiterChar = ',';
+ if (cmdLine.hasOption(DELIMITER_OPT.getOpt())) {
+ String delimString = cmdLine.getOptionValue(DELIMITER_OPT.getOpt());
+ if (delimString.length() != 1) {
+ throw new IllegalArgumentException("Illegal delimiter character: " + delimString);
+ }
+ delimiterChar = delimString.charAt(0);
+ }
+
+ if (cmdLine.hasOption(ZK_QUORUM_OPT.getOpt())) {
+ String zkQuorum = cmdLine.getOptionValue(ZK_QUORUM_OPT.getOpt());
+ LOG.info("Configuring ZK quorum to {}", zkQuorum);
+ conf.set(HConstants.ZOOKEEPER_QUORUM, zkQuorum);
+ }
+
+ CsvBulkImportUtil.initCsvImportJob(
+ conf,
+ getQualifiedTableName(
+ cmdLine.getOptionValue(SCHEMA_NAME_OPT.getOpt()),
+ cmdLine.getOptionValue(TABLE_NAME_OPT.getOpt())),
+ delimiterChar,
+ cmdLine.getOptionValue(ARRAY_DELIMITER_OPT.getOpt()),
+ importColumns,
+ cmdLine.hasOption(IGNORE_ERRORS_OPT.getOpt()));
+ }
+
+ /**
+ * Perform any required validation on the table being bulk loaded into:
+ * - ensure no column family names start with '_', as they'd be ignored leading to problems.
+ * @throws java.sql.SQLException
+ */
+ private void validateTable(Connection conn, String schemaName,
+ String tableName) throws SQLException {
+
+ ResultSet rs = conn.getMetaData().getColumns(
+ null, StringUtil.escapeLike(schemaName),
+ StringUtil.escapeLike(tableName), null);
+ while (rs.next()) {
+ String familyName = rs.getString(PhoenixDatabaseMetaData.COLUMN_FAMILY);
+ if (familyName != null && familyName.startsWith("_")) {
+ if (QueryConstants.DEFAULT_COLUMN_FAMILY.equals(familyName)) {
+ throw new IllegalStateException(
+ "CSV Bulk Loader error: All column names that are not part of the " +
+ "primary key constraint must be prefixed with a column family " +
+ "name (i.e. f.my_column VARCHAR)");
+ } else {
+ throw new IllegalStateException("CSV Bulk Loader error: Column family name " +
+ "must not start with '_': " + familyName);
+ }
+ }
+ }
+ rs.close();
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/ed9f58ba/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/CsvToKeyValueMapper.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/CsvToKeyValueMapper.java b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/CsvToKeyValueMapper.java
new file mode 100644
index 0000000..1d8dbf7
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/CsvToKeyValueMapper.java
@@ -0,0 +1,307 @@
+/*
+ * 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.
+ */
+package org.apache.phoenix.mapreduce;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Function;
+import com.google.common.base.Joiner;
+import com.google.common.base.Preconditions;
+import com.google.common.base.Splitter;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Iterables;
+import com.google.common.collect.Lists;
+import org.apache.commons.csv.CSVFormat;
+import org.apache.commons.csv.CSVParser;
+import org.apache.commons.csv.CSVRecord;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.HConstants;
+import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
+import org.apache.hadoop.hbase.util.Pair;
+import org.apache.hadoop.io.LongWritable;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.mapreduce.Mapper;
+import org.apache.hadoop.util.ReflectionUtils;
+import org.apache.phoenix.hbase.index.util.ClientKeyValue;
+import org.apache.phoenix.jdbc.PhoenixConnection;
+import org.apache.phoenix.jdbc.PhoenixDriver;
+import org.apache.phoenix.util.ColumnInfo;
+import org.apache.phoenix.util.PhoenixRuntime;
+import org.apache.phoenix.util.csv.CsvUpsertExecutor;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+import java.io.IOException;
+import java.io.StringReader;
+import java.sql.DriverManager;
+import java.sql.SQLException;
+import java.util.Iterator;
+import java.util.List;
+
+/**
+ * MapReduce mapper that converts CSV input lines into KeyValues that can be written to HFiles.
+ * <p/>
+ * KeyValues are produced by executing UPSERT statements on a Phoenix connection and then
+ * extracting the created KeyValues and rolling back the statement execution before it is
+ * committed to HBase.
+ */
+public class CsvToKeyValueMapper extends Mapper<LongWritable,Text,ImmutableBytesWritable,
+ KeyValue> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(CsvToKeyValueMapper.class);
+
+ private static final String COUNTER_GROUP_NAME = "Phoenix MapReduce Import";
+
+ /** Configuration key for the class name of an ImportPreUpsertKeyValueProcessor */
+ public static final String UPSERT_HOOK_CLASS_CONFKEY = "phoenix.mapreduce.import.kvprocessor";
+
+ /** Configuration key for the field delimiter for input csv records */
+ public static final String FIELD_DELIMITER_CONFKEY = "phoenix.mapreduce.import.fielddelimiter";
+
+ /** Configuration key for the array element delimiter for input arrays */
+ public static final String ARRAY_DELIMITER_CONFKEY = "phoenix.mapreduce.import.arraydelimiter";
+
+ /** Configuration key for the name of the output table */
+ public static final String TABLE_NAME_CONFKEY = "phoenix.mapreduce.import.tablename";
+
+ /** Configuration key for the columns to be imported */
+ public static final String COLUMN_INFO_CONFKEY = "phoenix.mapreduce.import.columninfos";
+
+ /** Configuration key for the flag to ignore invalid rows */
+ public static final String IGNORE_INVALID_ROW_CONFKEY = "phoenix.mapreduce.import.ignoreinvalidrow";
+
+ private PhoenixConnection conn;
+ private CsvUpsertExecutor csvUpsertExecutor;
+ private MapperUpsertListener upsertListener;
+ private CsvLineParser csvLineParser;
+ private ImportPreUpsertKeyValueProcessor preUpdateProcessor;
+
+ @Override
+ protected void setup(Context context) throws IOException, InterruptedException {
+
+ Configuration conf = context.getConfiguration();
+ String jdbcUrl = getJdbcUrl(conf);
+
+ // This statement also ensures that the driver class is loaded
+ LOG.info("Connection with driver {} with url {}", PhoenixDriver.class.getName(), jdbcUrl);
+
+ try {
+ conn = (PhoenixConnection) DriverManager.getConnection(jdbcUrl);
+ } catch (SQLException e) {
+ throw new RuntimeException(e);
+ }
+
+ upsertListener = new MapperUpsertListener(
+ context, conf.getBoolean(IGNORE_INVALID_ROW_CONFKEY, true));
+ csvUpsertExecutor = buildUpsertExecutor(conf);
+ csvLineParser = new CsvLineParser(conf.get(FIELD_DELIMITER_CONFKEY).charAt(0));
+
+ preUpdateProcessor = loadPreUpsertProcessor(conf);
+ }
+
+ @Override
+ protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
+ ImmutableBytesWritable outputKey = new ImmutableBytesWritable();
+ try {
+ CSVRecord csvRecord = null;
+ try {
+ csvRecord = csvLineParser.parse(value.toString());
+ } catch (IOException e) {
+ context.getCounter(COUNTER_GROUP_NAME, "CSV Parser errors").increment(1L);
+ }
+
+ if (csvRecord == null) {
+ context.getCounter(COUNTER_GROUP_NAME, "Empty records").increment(1L);
+ return;
+ }
+ csvUpsertExecutor.execute(ImmutableList.of(csvRecord));
+
+ Iterator<Pair<byte[], List<KeyValue>>> uncommittedDataIterator
+ = PhoenixRuntime.getUncommittedDataIterator(conn);
+ while (uncommittedDataIterator.hasNext()) {
+ Pair<byte[], List<KeyValue>> kvPair = uncommittedDataIterator.next();
+ List<KeyValue> keyValueList = kvPair.getSecond();
+
+ keyValueList = preUpdateProcessor.preUpsert(kvPair.getFirst(), keyValueList);
+ for (KeyValue kv : keyValueList) {
+ if (kv instanceof ClientKeyValue) {
+ kv = new KeyValue(kv.getRow(), kv.getFamily(), kv.getQualifier(),
+ kv.getTimestamp(), kv.getValue());
+ }
+ outputKey.set(kv.getBuffer(), kv.getRowOffset(), kv.getRowLength());
+ context.write(outputKey, kv);
+ }
+ }
+ conn.rollback();
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ protected void cleanup(Context context) throws IOException, InterruptedException {
+ try {
+ conn.close();
+ } catch (SQLException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ /**
+ * Load the configured ImportPreUpsertKeyValueProcessor, or supply a dummy processor.
+ */
+ @VisibleForTesting
+ static ImportPreUpsertKeyValueProcessor loadPreUpsertProcessor(Configuration conf) {
+ Class<? extends ImportPreUpsertKeyValueProcessor> processorClass = null;
+ try {
+ processorClass = conf.getClass(
+ UPSERT_HOOK_CLASS_CONFKEY, DefaultImportPreUpsertKeyValueProcessor.class,
+ ImportPreUpsertKeyValueProcessor.class);
+ } catch (Exception e) {
+ throw new IllegalStateException("Couldn't load upsert hook class", e);
+ }
+
+ return ReflectionUtils.newInstance(processorClass, conf);
+ }
+
+ /**
+ * Build up the JDBC URL for connecting to Phoenix.
+ *
+ * @return the full JDBC URL for a Phoenix connection
+ */
+ @VisibleForTesting
+ static String getJdbcUrl(Configuration conf) {
+ String zkQuorum = conf.get(HConstants.ZOOKEEPER_QUORUM);
+ if (zkQuorum == null) {
+ throw new IllegalStateException(HConstants.ZOOKEEPER_QUORUM + " is not configured");
+ }
+ return PhoenixRuntime.JDBC_PROTOCOL + PhoenixRuntime.JDBC_PROTOCOL_SEPARATOR + zkQuorum;
+ }
+
+ @VisibleForTesting
+ CsvUpsertExecutor buildUpsertExecutor(Configuration conf) {
+ String tableName = conf.get(TABLE_NAME_CONFKEY);
+ String arraySeparator = conf.get(ARRAY_DELIMITER_CONFKEY);
+ Preconditions.checkNotNull(tableName, "table name is not configured");
+
+ List<ColumnInfo> columnInfoList = buildColumnInfoList(conf);
+
+ return CsvUpsertExecutor.create(conn, tableName, columnInfoList, upsertListener, arraySeparator);
+ }
+
+ /**
+ * Write the list of to-import columns to a job configuration.
+ *
+ * @param conf configuration to be written to
+ * @param columnInfoList list of ColumnInfo objects to be configured for import
+ */
+ @VisibleForTesting
+ static void configureColumnInfoList(Configuration conf, List<ColumnInfo> columnInfoList) {
+ conf.set(COLUMN_INFO_CONFKEY, Joiner.on("|").useForNull("").join(columnInfoList));
+ }
+
+ /**
+ * Build the list of ColumnInfos for the import based on information in the configuration.
+ */
+ @VisibleForTesting
+ static List<ColumnInfo> buildColumnInfoList(Configuration conf) {
+
+ return Lists.newArrayList(
+ Iterables.transform(
+ Splitter.on("|").split(conf.get(COLUMN_INFO_CONFKEY)),
+ new Function<String, ColumnInfo>() {
+ @Nullable
+ @Override
+ public ColumnInfo apply(@Nullable String input) {
+ if (input.isEmpty()) {
+ // An empty string represents a null that was passed in to
+ // the configuration, which corresponds to an input column
+ // which is to be skipped
+ return null;
+ }
+ return ColumnInfo.fromString(input);
+ }
+ }));
+ }
+
+ /**
+ * Listener that logs successful upserts and errors to job counters.
+ */
+ @VisibleForTesting
+ static class MapperUpsertListener implements CsvUpsertExecutor.UpsertListener {
+
+ private final Context context;
+ private final boolean ignoreRecordErrors;
+
+ private MapperUpsertListener(Context context, boolean ignoreRecordErrors) {
+ this.context = context;
+ this.ignoreRecordErrors = ignoreRecordErrors;
+ }
+
+ @Override
+ public void upsertDone(long upsertCount) {
+ context.getCounter(COUNTER_GROUP_NAME, "Upserts Done").increment(1L);
+ }
+
+ @Override
+ public void errorOnRecord(CSVRecord csvRecord, String errorMessage) {
+ LOG.error("Error on record {}: {}", csvRecord, errorMessage);
+ context.getCounter(COUNTER_GROUP_NAME, "Errors on records").increment(1L);
+ if (!ignoreRecordErrors) {
+ throw new RuntimeException("Error on record, " + errorMessage + ", " +
+ "record =" + csvRecord);
+ }
+ }
+ }
+
+ /**
+ * Parses a single CSV input line, returning a {@code CSVRecord}.
+ */
+ @VisibleForTesting
+ static class CsvLineParser {
+
+ private final CSVFormat csvFormat;
+
+ CsvLineParser(char fieldDelimiter) {
+ this.csvFormat = CSVFormat.newFormat(fieldDelimiter);
+ }
+
+ public CSVRecord parse(String input) throws IOException {
+ // TODO Creating a new parser for each line seems terribly inefficient but
+ // there's no public way to parse single lines via commons-csv. We should update
+ // it to create a LineParser class like this one.
+ CSVParser csvParser = new CSVParser(new StringReader(input), csvFormat);
+ return Iterables.getFirst(csvParser, null);
+ }
+ }
+
+ /**
+ * A default implementation of {@code ImportPreUpsertKeyValueProcessor} that is used if no
+ * specific class is configured. This implementation simply passes through the KeyValue
+ * list that is passed in.
+ */
+ public static class DefaultImportPreUpsertKeyValueProcessor implements
+ ImportPreUpsertKeyValueProcessor {
+
+ @Override
+ public List<KeyValue> preUpsert(byte[] rowKey, List<KeyValue> keyValues) {
+ return keyValues;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/ed9f58ba/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/ImportPreUpsertKeyValueProcessor.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/ImportPreUpsertKeyValueProcessor.java b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/ImportPreUpsertKeyValueProcessor.java
new file mode 100644
index 0000000..62211f3
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/ImportPreUpsertKeyValueProcessor.java
@@ -0,0 +1,47 @@
+/*
+ * 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.
+ */
+package org.apache.phoenix.mapreduce;
+
+import org.apache.hadoop.hbase.KeyValue;
+
+import java.util.List;
+
+/**
+ * A listener hook to process KeyValues that are being written to HFiles for bulk import.
+ * Implementing this interface and configuring it via the {@link
+ * CsvToKeyValueMapper#UPSERT_HOOK_CLASS_CONFKEY} configuration key.
+ * <p/>
+ * The intention of such a hook is to allow coproccessor-style operations to be peformed on
+ * data that is being bulk-loaded via MapReduce.
+ */
+public interface ImportPreUpsertKeyValueProcessor {
+
+ /**
+ * Process a list of KeyValues before they are written to an HFile. The supplied list of
+ * KeyValues contain all data that is to be written for a single Phoenix row.
+ * <p/>
+ * Implementors can filter certain KeyValues from the list, augment the list, or return the
+ * same list.
+ *
+ * @param rowKey the row key for the key values that are being passed in
+ * @param keyValues list of KeyValues that are to be written to an HFile
+ * @return the list that will actually be written
+ */
+ List<KeyValue> preUpsert(byte[] rowKey, List<KeyValue> keyValues);
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/ed9f58ba/phoenix-core/src/main/java/org/apache/phoenix/schema/PDataType.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/schema/PDataType.java b/phoenix-core/src/main/java/org/apache/phoenix/schema/PDataType.java
index 34166d3..ee1e288 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/schema/PDataType.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/schema/PDataType.java
@@ -6965,7 +6965,7 @@ public enum PDataType {
}
}
}
- public static PDataType fromTypeId(Integer typeId) {
+ public static PDataType fromTypeId(int typeId) {
int offset = typeId - SQL_TYPE_OFFSET;
if (offset >= 0 && offset < SQL_TYPE_TO_PCOLUMN_DATA_TYPE.length) {
PDataType type = SQL_TYPE_TO_PCOLUMN_DATA_TYPE[offset];
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/ed9f58ba/phoenix-core/src/main/java/org/apache/phoenix/util/CSVCommonsLoader.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/util/CSVCommonsLoader.java b/phoenix-core/src/main/java/org/apache/phoenix/util/CSVCommonsLoader.java
index 3e86477..0292aae 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/util/CSVCommonsLoader.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/util/CSVCommonsLoader.java
@@ -19,6 +19,7 @@ package org.apache.phoenix.util;
import java.io.File;
import java.io.Reader;
+import java.sql.Connection;
import java.sql.DatabaseMetaData;
import java.sql.ResultSet;
import java.sql.SQLException;
@@ -72,7 +73,6 @@ public class CSVCommonsLoader {
put("9",'\u0009');}};
private final String arrayElementSeparator;
- private int unfoundColumnCount;
public enum PhoenixHeaderSource {
FROM_TABLE,
@@ -269,21 +269,26 @@ public class CSVCommonsLoader {
default:
throw new IllegalStateException("parser has unknown column source.");
}
- return generateColumnInfo(columns);
+ return generateColumnInfo(conn, tableName, columns, isStrict);
}
/**
- * Get array of ColumnInfos that contain Column Name and its associated
- * PDataType
+ * Get list of ColumnInfos that contain Column Name and its associated
+ * PDataType for an import. The supplied list of columns can be null -- if it is non-null,
+ * it represents a user-supplied list of columns to be imported.
*
- * @param columns
- * @return
- * @throws SQLException
+ * @param conn Phoenix connection from which metadata will be read
+ * @param tableName Phoenix table name whose columns are to be checked. Can include a schema
+ * name
+ * @param columns user-supplied list of import columns, can be null
+ * @param strict if true, an exception will be thrown if unknown columns are supplied
*/
- private List<ColumnInfo> generateColumnInfo(List<String> columns)
+ public static List<ColumnInfo> generateColumnInfo(Connection conn,
+ String tableName, List<String> columns, boolean strict)
throws SQLException {
Map<String, Integer> columnNameToTypeMap = Maps.newLinkedHashMap();
DatabaseMetaData dbmd = conn.getMetaData();
+ int unfoundColumnCount = 0;
// TODO: escape wildcard characters here because we don't want that
// behavior here
String escapedTableName = StringUtil.escapeLike(tableName);
@@ -317,7 +322,7 @@ public class CSVCommonsLoader {
String columnName = columns.get(i).trim();
Integer sqlType = columnNameToTypeMap.get(columnName);
if (sqlType == null) {
- if (isStrict) {
+ if (strict) {
throw new SQLExceptionInfo.Builder(
SQLExceptionCode.COLUMN_NOT_FOUND)
.setColumnName(columnName)
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/ed9f58ba/phoenix-core/src/main/java/org/apache/phoenix/util/ColumnInfo.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/util/ColumnInfo.java b/phoenix-core/src/main/java/org/apache/phoenix/util/ColumnInfo.java
index 37238c8..2ebc07f 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/util/ColumnInfo.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/util/ColumnInfo.java
@@ -10,16 +10,30 @@
package org.apache.phoenix.util;
+import com.google.common.base.Joiner;
+import com.google.common.base.Preconditions;
+import com.google.common.base.Splitter;
+import com.google.common.collect.Lists;
+import org.apache.phoenix.schema.IllegalDataException;
import org.apache.phoenix.schema.PDataType;
+import java.util.ArrayList;
+import java.util.List;
+
/**
* ColumnInfo used to store Column Name and its associated PDataType
*/
public class ColumnInfo {
- private String columnName;
- private Integer sqlType;
- public ColumnInfo(String columnName, Integer sqlType) {
+ /** Separator used for the toString representation */
+ private static final String STR_SEPARATOR = ":";
+
+ private final String columnName;
+ private final int sqlType;
+
+ public ColumnInfo(String columnName, int sqlType) {
+ Preconditions.checkNotNull(columnName, "columnName cannot be null");
+ Preconditions.checkArgument(!columnName.isEmpty(), "columnName cannot be empty");
this.columnName = columnName;
this.sqlType = sqlType;
}
@@ -28,12 +42,58 @@ public class ColumnInfo {
return columnName;
}
- public Integer getSqlType() {
+ public int getSqlType() {
return sqlType;
}
+ public PDataType getPDataType() {
+ return PDataType.fromTypeId(sqlType);
+ }
+
@Override
public String toString() {
- return columnName + " " + PDataType.fromTypeId(sqlType).getSqlTypeName();
+ return columnName + STR_SEPARATOR + getPDataType().getSqlTypeName();
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+
+ ColumnInfo that = (ColumnInfo) o;
+
+ if (sqlType != that.sqlType) return false;
+ if (!columnName.equals(that.columnName)) return false;
+
+ return true;
+ }
+
+ @Override
+ public int hashCode() {
+ int result = columnName.hashCode();
+ result = 31 * result + sqlType;
+ return result;
}
+
+ /**
+ * Instantiate a {@code ColumnInfo} from a string representation created by the {@link
+ * #toString()} method.
+ *
+ * @param stringRepresentation string representation of a ColumnInfo
+ * @return the corresponding ColumnInfo
+ * @throws java.lang.IllegalArgumentException if the given string representation cannot be
+ * parsed
+ */
+ public static ColumnInfo fromString(String stringRepresentation) {
+ List<String> components =
+ Lists.newArrayList(Splitter.on(STR_SEPARATOR).split(stringRepresentation));
+ if (components.size() != 2) {
+ throw new IllegalArgumentException("Unparseable string: " + stringRepresentation);
+ }
+
+ return new ColumnInfo(
+ components.get(0),
+ PDataType.fromSqlTypeName(components.get(1)).getSqlType());
+ }
+
}
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/ed9f58ba/phoenix-core/src/test/java/org/apache/phoenix/mapreduce/CsvBulkImportUtilTest.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/test/java/org/apache/phoenix/mapreduce/CsvBulkImportUtilTest.java b/phoenix-core/src/test/java/org/apache/phoenix/mapreduce/CsvBulkImportUtilTest.java
new file mode 100644
index 0000000..821892d
--- /dev/null
+++ b/phoenix-core/src/test/java/org/apache/phoenix/mapreduce/CsvBulkImportUtilTest.java
@@ -0,0 +1,71 @@
+/*
+ * 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.
+ */
+package org.apache.phoenix.mapreduce;
+
+import com.google.common.collect.ImmutableList;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.KeyValue;
+import org.apache.phoenix.schema.PDataType;
+import org.apache.phoenix.util.ColumnInfo;
+import org.junit.Test;
+
+import java.util.List;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.fail;
+
+public class CsvBulkImportUtilTest {
+
+ @Test
+ public void testInitCsvImportJob() {
+ Configuration conf = new Configuration();
+
+ String tableName = "SCHEMANAME.TABLENAME";
+ char delimiter = '!';
+
+ List<ColumnInfo> columnInfoList = ImmutableList.of(
+ new ColumnInfo("MYCOL", PDataType.INTEGER.getSqlType()));
+
+ CsvBulkImportUtil.initCsvImportJob(
+ conf, tableName, delimiter, null, columnInfoList, true);
+
+ assertEquals(tableName, conf.get(CsvToKeyValueMapper.TABLE_NAME_CONFKEY));
+ assertEquals("!", conf.get(CsvToKeyValueMapper.FIELD_DELIMITER_CONFKEY));
+ assertNull(conf.get(CsvToKeyValueMapper.ARRAY_DELIMITER_CONFKEY));
+ assertEquals(columnInfoList, CsvToKeyValueMapper.buildColumnInfoList(conf));
+ assertEquals(true, conf.getBoolean(CsvToKeyValueMapper.IGNORE_INVALID_ROW_CONFKEY, false));
+ }
+
+ @Test
+ public void testConfigurePreUpsertProcessor() {
+ Configuration conf = new Configuration();
+ CsvBulkImportUtil.configurePreUpsertProcessor(conf, MockProcessor.class);
+ ImportPreUpsertKeyValueProcessor processor = CsvToKeyValueMapper.loadPreUpsertProcessor(conf);
+ assertEquals(MockProcessor.class, processor.getClass());
+ }
+
+
+ public static class MockProcessor implements ImportPreUpsertKeyValueProcessor {
+
+ @Override
+ public List<KeyValue> preUpsert(byte[] rowKey, List<KeyValue> keyValues) {
+ throw new UnsupportedOperationException("Not yet implemented");
+ }
+ }
+}