You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@phoenix.apache.org by nd...@apache.org on 2015/12/08 00:02:05 UTC
[1/3] phoenix git commit: PHOENIX-2481 JSON bulkload tool
Repository: phoenix
Updated Branches:
refs/heads/master 8ce3b580f -> 578979a14
http://git-wip-us.apache.org/repos/asf/phoenix/blob/578979a1/phoenix-core/src/test/java/org/apache/phoenix/mapreduce/FormatToKeyValueMapperTest.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/test/java/org/apache/phoenix/mapreduce/FormatToKeyValueMapperTest.java b/phoenix-core/src/test/java/org/apache/phoenix/mapreduce/FormatToKeyValueMapperTest.java
new file mode 100644
index 0000000..3455616
--- /dev/null
+++ b/phoenix-core/src/test/java/org/apache/phoenix/mapreduce/FormatToKeyValueMapperTest.java
@@ -0,0 +1,102 @@
+/*
+ * 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 java.util.List;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.KeyValue;
+import org.apache.phoenix.mapreduce.util.PhoenixConfigurationUtil;
+import org.apache.phoenix.schema.types.PInteger;
+import org.apache.phoenix.schema.types.PIntegerArray;
+import org.apache.phoenix.schema.types.PUnsignedInt;
+import org.apache.phoenix.util.ColumnInfo;
+import org.junit.Test;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Lists;
+
+import static org.junit.Assert.assertEquals;
+
+public class FormatToKeyValueMapperTest {
+
+ @Test
+ public void testBuildColumnInfoList() {
+ List<ColumnInfo> columnInfoList = ImmutableList.of(
+ new ColumnInfo("idCol", PInteger.INSTANCE.getSqlType()),
+ new ColumnInfo("unsignedIntCol", PUnsignedInt.INSTANCE.getSqlType()),
+ new ColumnInfo("stringArrayCol", PIntegerArray.INSTANCE.getSqlType()));
+
+ Configuration conf = new Configuration();
+ FormatToKeyValueMapper.configureColumnInfoList(conf, columnInfoList);
+ List<ColumnInfo> fromConfig = FormatToKeyValueMapper.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", PInteger.INSTANCE.getSqlType()),
+ null,
+ new ColumnInfo("unsignedIntCol", PUnsignedInt.INSTANCE.getSqlType()),
+ new ColumnInfo("stringArrayCol", PIntegerArray.INSTANCE.getSqlType()));
+
+ Configuration conf = new Configuration();
+ FormatToKeyValueMapper.configureColumnInfoList(conf, columnInfoListWithNull);
+ List<ColumnInfo> fromConfig = FormatToKeyValueMapper.buildColumnInfoList(conf);
+
+ assertEquals(columnInfoListWithNull, fromConfig);
+ }
+
+ @Test
+ public void testLoadPreUpdateProcessor() {
+ Configuration conf = new Configuration();
+ conf.setClass(PhoenixConfigurationUtil.UPSERT_HOOK_CLASS_CONFKEY, MockUpsertProcessor.class,
+ ImportPreUpsertKeyValueProcessor.class);
+
+ ImportPreUpsertKeyValueProcessor processor = PhoenixConfigurationUtil.loadPreUpsertProcessor(conf);
+ assertEquals(MockUpsertProcessor.class, processor.getClass());
+ }
+
+ @Test
+ public void testLoadPreUpdateProcessor_NotConfigured() {
+
+ Configuration conf = new Configuration();
+ ImportPreUpsertKeyValueProcessor processor = PhoenixConfigurationUtil.loadPreUpsertProcessor(conf);
+
+ assertEquals(FormatToKeyValueMapper.DefaultImportPreUpsertKeyValueProcessor.class,
+ processor.getClass());
+ }
+
+ @Test(expected=IllegalStateException.class)
+ public void testLoadPreUpdateProcessor_ClassNotFound() {
+ Configuration conf = new Configuration();
+ conf.set(PhoenixConfigurationUtil.UPSERT_HOOK_CLASS_CONFKEY, "MyUndefinedClass");
+
+ PhoenixConfigurationUtil.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/phoenix/blob/578979a1/phoenix-core/src/test/java/org/apache/phoenix/util/AbstractUpsertExecutorTest.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/test/java/org/apache/phoenix/util/AbstractUpsertExecutorTest.java b/phoenix-core/src/test/java/org/apache/phoenix/util/AbstractUpsertExecutorTest.java
new file mode 100644
index 0000000..b614312
--- /dev/null
+++ b/phoenix-core/src/test/java/org/apache/phoenix/util/AbstractUpsertExecutorTest.java
@@ -0,0 +1,136 @@
+/*
+ * 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 java.io.IOException;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.sql.Types;
+import java.util.Arrays;
+import java.util.List;
+
+import org.apache.phoenix.query.BaseConnectionlessQueryTest;
+import org.apache.phoenix.schema.types.PArrayDataType;
+import org.apache.phoenix.schema.types.PInteger;
+import org.apache.phoenix.schema.types.PIntegerArray;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import com.google.common.collect.ImmutableList;
+
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoMoreInteractions;
+
+public abstract class AbstractUpsertExecutorTest<R, F> extends BaseConnectionlessQueryTest {
+
+ protected Connection conn;
+ protected List<ColumnInfo> columnInfoList;
+ protected PreparedStatement preparedStatement;
+ protected UpsertExecutor.UpsertListener<R> upsertListener;
+
+ protected abstract UpsertExecutor<R, F> getUpsertExecutor();
+ protected abstract R createRecord(Object... columnValues) throws IOException;
+
+ @Before
+ public void setUp() throws SQLException {
+ columnInfoList = ImmutableList.of(
+ new ColumnInfo("ID", Types.BIGINT),
+ new ColumnInfo("NAME", Types.VARCHAR),
+ new ColumnInfo("AGE", Types.INTEGER),
+ new ColumnInfo("VALUES", PIntegerArray.INSTANCE.getSqlType()));
+
+ preparedStatement = mock(PreparedStatement.class);
+ upsertListener = mock(UpsertExecutor.UpsertListener.class);
+ conn = DriverManager.getConnection(getUrl());
+ }
+
+ @After
+ public void tearDown() throws SQLException {
+ conn.close();
+ }
+
+ @Test
+ public void testExecute() throws Exception {
+ getUpsertExecutor().execute(createRecord(123L, "NameValue", 42, Arrays.asList(1, 2, 3)));
+
+ verify(upsertListener).upsertDone(1L);
+ verifyNoMoreInteractions(upsertListener);
+
+ verify(preparedStatement).setObject(1, Long.valueOf(123L));
+ verify(preparedStatement).setObject(2, "NameValue");
+ verify(preparedStatement).setObject(3, Integer.valueOf(42));
+ verify(preparedStatement).setObject(4, PArrayDataType.instantiatePhoenixArray(PInteger.INSTANCE, new Object[]{1,2,3}));
+ verify(preparedStatement).execute();
+ verifyNoMoreInteractions(preparedStatement);
+ }
+
+ @Test
+ public void testExecute_TooFewFields() throws Exception {
+ R recordWithTooFewFields = createRecord(123L, "NameValue");
+ getUpsertExecutor().execute(recordWithTooFewFields);
+
+ verify(upsertListener).errorOnRecord(eq(recordWithTooFewFields), any(Throwable.class));
+ verifyNoMoreInteractions(upsertListener);
+ }
+
+ @Test
+ public void testExecute_TooManyFields() throws Exception {
+ R recordWithTooManyFields = createRecord(123L, "NameValue", 42, Arrays.asList(1, 2, 3), "Garbage");
+ getUpsertExecutor().execute(recordWithTooManyFields);
+
+ verify(upsertListener).upsertDone(1L);
+ verifyNoMoreInteractions(upsertListener);
+
+ verify(preparedStatement).setObject(1, Long.valueOf(123L));
+ verify(preparedStatement).setObject(2, "NameValue");
+ verify(preparedStatement).setObject(3, Integer.valueOf(42));
+ verify(preparedStatement).setObject(4, PArrayDataType.instantiatePhoenixArray(PInteger.INSTANCE, new Object[]{1,2,3}));
+ verify(preparedStatement).execute();
+ verifyNoMoreInteractions(preparedStatement);
+ }
+
+ @Test
+ public void testExecute_NullField() throws Exception {
+ getUpsertExecutor().execute(createRecord(123L, "NameValue", null, Arrays.asList(1, 2, 3)));
+
+ verify(upsertListener).upsertDone(1L);
+ verifyNoMoreInteractions(upsertListener);
+
+ verify(preparedStatement).setObject(1, Long.valueOf(123L));
+ verify(preparedStatement).setObject(2, "NameValue");
+ verify(preparedStatement).setNull(3, columnInfoList.get(2).getSqlType());
+ verify(preparedStatement).setObject(4, PArrayDataType.instantiatePhoenixArray(PInteger.INSTANCE, new Object[]{1,2,3}));
+ verify(preparedStatement).execute();
+ verifyNoMoreInteractions(preparedStatement);
+ }
+
+ @Test
+ public void testExecute_InvalidType() throws Exception {
+ R recordWithInvalidType = createRecord(123L, "NameValue", "ThisIsNotANumber", Arrays.asList(1, 2, 3));
+ getUpsertExecutor().execute(recordWithInvalidType);
+
+ verify(upsertListener).errorOnRecord(eq(recordWithInvalidType), any(Throwable.class));
+ verifyNoMoreInteractions(upsertListener);
+ }
+}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/578979a1/phoenix-core/src/test/java/org/apache/phoenix/util/csv/CsvUpsertExecutorTest.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/test/java/org/apache/phoenix/util/csv/CsvUpsertExecutorTest.java b/phoenix-core/src/test/java/org/apache/phoenix/util/csv/CsvUpsertExecutorTest.java
index 6efe246..7a09bee 100644
--- a/phoenix-core/src/test/java/org/apache/phoenix/util/csv/CsvUpsertExecutorTest.java
+++ b/phoenix-core/src/test/java/org/apache/phoenix/util/csv/CsvUpsertExecutorTest.java
@@ -17,130 +17,50 @@
*/
package org.apache.phoenix.util.csv;
-import com.google.common.base.Joiner;
-import com.google.common.collect.ImmutableList;
-import com.google.common.collect.Iterables;
-import org.apache.commons.csv.CSVFormat;
-import org.apache.commons.csv.CSVParser;
-import org.apache.commons.csv.CSVRecord;
-import org.apache.phoenix.query.BaseConnectionlessQueryTest;
-import org.apache.phoenix.schema.types.PInteger;
-import org.apache.phoenix.schema.types.PIntegerArray;
-import org.apache.phoenix.schema.types.PArrayDataType;
-import org.apache.phoenix.util.ColumnInfo;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.Test;
-
import java.io.IOException;
-import java.sql.Connection;
-import java.sql.DriverManager;
-import java.sql.PreparedStatement;
import java.sql.SQLException;
-import java.sql.Types;
import java.util.List;
-import static org.mockito.Matchers.any;
-import static org.mockito.Matchers.anyString;
-import static org.mockito.Matchers.eq;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.verifyNoMoreInteractions;
-
-public class CsvUpsertExecutorTest extends BaseConnectionlessQueryTest {
-
- private Connection conn;
- private List<ColumnInfo> columnInfoList;
- private PreparedStatement preparedStatement;
- private CsvUpsertExecutor.UpsertListener upsertListener;
-
- private CsvUpsertExecutor upsertExecutor;
-
- @Before
- public void setUp() throws SQLException {
- columnInfoList = ImmutableList.of(
- new ColumnInfo("ID", Types.BIGINT),
- new ColumnInfo("NAME", Types.VARCHAR),
- new ColumnInfo("AGE", Types.INTEGER),
- new ColumnInfo("VALUES", PIntegerArray.INSTANCE.getSqlType()));
-
- preparedStatement = mock(PreparedStatement.class);
- upsertListener = mock(CsvUpsertExecutor.UpsertListener.class);
- conn = DriverManager.getConnection(getUrl());
- upsertExecutor = new CsvUpsertExecutor(conn, columnInfoList, preparedStatement, upsertListener, ":");
- }
-
- @After
- public void tearDown() throws SQLException {
- conn.close();
- }
-
- @Test
- public void testExecute() throws Exception {
- upsertExecutor.execute(createCsvRecord("123,NameValue,42,1:2:3"));
-
- verify(upsertListener).upsertDone(1L);
- verifyNoMoreInteractions(upsertListener);
-
- verify(preparedStatement).setObject(1, Long.valueOf(123L));
- verify(preparedStatement).setObject(2, "NameValue");
- verify(preparedStatement).setObject(3, Integer.valueOf(42));
- verify(preparedStatement).setObject(4, PArrayDataType.instantiatePhoenixArray(PInteger.INSTANCE, new Object[]{1,2,3}));
- verify(preparedStatement).execute();
- verifyNoMoreInteractions(preparedStatement);
- }
-
- @Test
- public void testExecute_TooFewFields() throws Exception {
- CSVRecord csvRecordWithTooFewFields = createCsvRecord("123,NameValue");
- upsertExecutor.execute(csvRecordWithTooFewFields);
-
- verify(upsertListener).errorOnRecord(eq(csvRecordWithTooFewFields), any(Throwable.class));
- verifyNoMoreInteractions(upsertListener);
- }
-
- @Test
- public void testExecute_TooManyFields() throws Exception {
- CSVRecord csvRecordWithTooManyFields = createCsvRecord("123,NameValue,42,1:2:3,Garbage");
- upsertExecutor.execute(csvRecordWithTooManyFields);
-
- verify(upsertListener).upsertDone(1L);
- verifyNoMoreInteractions(upsertListener);
-
- verify(preparedStatement).setObject(1, Long.valueOf(123L));
- verify(preparedStatement).setObject(2, "NameValue");
- verify(preparedStatement).setObject(3, Integer.valueOf(42));
- verify(preparedStatement).setObject(4, PArrayDataType.instantiatePhoenixArray(PInteger.INSTANCE, new Object[]{1,2,3}));
- verify(preparedStatement).execute();
- verifyNoMoreInteractions(preparedStatement);
- }
+import org.apache.commons.csv.CSVFormat;
+import org.apache.commons.csv.CSVParser;
+import org.apache.commons.csv.CSVRecord;
+import org.apache.phoenix.util.AbstractUpsertExecutorTest;
+import org.apache.phoenix.util.UpsertExecutor;
+import org.junit.Before;
- @Test
- public void testExecute_NullField() throws Exception {
- upsertExecutor.execute(createCsvRecord("123,NameValue,,1:2:3"));
+import com.google.common.base.Joiner;
+import com.google.common.collect.Iterables;
- verify(upsertListener).upsertDone(1L);
- verifyNoMoreInteractions(upsertListener);
+public class CsvUpsertExecutorTest extends AbstractUpsertExecutorTest<CSVRecord, String> {
- verify(preparedStatement).setObject(1, Long.valueOf(123L));
- verify(preparedStatement).setObject(2, "NameValue");
- verify(preparedStatement).setNull(3, columnInfoList.get(2).getSqlType());
- verify(preparedStatement).setObject(4, PArrayDataType.instantiatePhoenixArray(PInteger.INSTANCE, new Object[]{1,2,3}));
- verify(preparedStatement).execute();
- verifyNoMoreInteractions(preparedStatement);
- }
+ private static final String ARRAY_SEP = ":";
- @Test
- public void testExecute_InvalidType() throws Exception {
- CSVRecord csvRecordWithInvalidType = createCsvRecord("123,NameValue,ThisIsNotANumber,1:2:3");
- upsertExecutor.execute(csvRecordWithInvalidType);
+ private UpsertExecutor<CSVRecord, String> upsertExecutor;
- verify(upsertListener).errorOnRecord(eq(csvRecordWithInvalidType), any(Throwable.class));
- verifyNoMoreInteractions(upsertListener);
+ @Override
+ public UpsertExecutor<CSVRecord, String> getUpsertExecutor() {
+ return upsertExecutor;
}
- private CSVRecord createCsvRecord(String...columnValues) throws IOException {
+ @Override
+ public CSVRecord createRecord(Object... columnValues) throws IOException {
+ for (int i = 0; i < columnValues.length; i++) {
+ if (columnValues[i] == null) {
+ // Joiner.join throws on nulls, replace with empty string.
+ columnValues[i] = "";
+ }
+ if (columnValues[i] instanceof List) {
+ columnValues[i] = Joiner.on(ARRAY_SEP).join((List<?>) columnValues[i]);
+ }
+ }
String inputRecord = Joiner.on(',').join(columnValues);
return Iterables.getFirst(CSVParser.parse(inputRecord, CSVFormat.DEFAULT), null);
}
+
+ @Before
+ public void setUp() throws SQLException {
+ super.setUp();
+ upsertExecutor = new CsvUpsertExecutor(conn, columnInfoList, preparedStatement,
+ upsertListener, ARRAY_SEP);
+ }
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/578979a1/phoenix-core/src/test/java/org/apache/phoenix/util/json/JsonUpsertExecutorTest.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/test/java/org/apache/phoenix/util/json/JsonUpsertExecutorTest.java b/phoenix-core/src/test/java/org/apache/phoenix/util/json/JsonUpsertExecutorTest.java
new file mode 100644
index 0000000..c042dd4
--- /dev/null
+++ b/phoenix-core/src/test/java/org/apache/phoenix/util/json/JsonUpsertExecutorTest.java
@@ -0,0 +1,53 @@
+/*
+ * 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.json;
+
+import java.io.IOException;
+import java.sql.SQLException;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.phoenix.util.AbstractUpsertExecutorTest;
+import org.apache.phoenix.util.UpsertExecutor;
+import org.junit.Before;
+
+public class JsonUpsertExecutorTest extends AbstractUpsertExecutorTest<Map<?, ?>, Object> {
+
+ private UpsertExecutor<Map<?, ?>, Object> upsertExecutor;
+
+ @Override
+ protected UpsertExecutor<Map<?, ?>, Object> getUpsertExecutor() {
+ return upsertExecutor;
+ }
+
+ @Override
+ protected Map<?, ?> createRecord(Object... columnValues) throws IOException {
+ Map ret = new HashMap(columnValues.length);
+ int min = Math.min(columnInfoList.size(), columnValues.length);
+ for (int i = 0; i < min; i++) {
+ ret.put(columnInfoList.get(i).getColumnName().replace("\"", "").toLowerCase(), columnValues[i]);
+ }
+ return ret;
+ }
+
+ @Before
+ public void setUp() throws SQLException {
+ super.setUp();
+ upsertExecutor = new JsonUpsertExecutor(conn, columnInfoList, preparedStatement, upsertListener);
+ }
+}
[3/3] phoenix git commit: PHOENIX-2481 JSON bulkload tool
Posted by nd...@apache.org.
PHOENIX-2481 JSON bulkload tool
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/578979a1
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/578979a1
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/578979a1
Branch: refs/heads/master
Commit: 578979a1437124d53e319ad554b72793bcef6fd3
Parents: 8ce3b58
Author: Nick Dimiduk <nd...@apache.org>
Authored: Mon Nov 16 17:18:34 2015 -0800
Committer: Nick Dimiduk <nd...@apache.org>
Committed: Mon Dec 7 13:09:09 2015 -0800
----------------------------------------------------------------------
.../phoenix/mapreduce/CsvBulkLoadToolIT.java | 16 +-
.../phoenix/mapreduce/AbstractBulkLoadTool.java | 402 +++++++++++++++
.../phoenix/mapreduce/CsvBulkImportUtil.java | 20 +-
.../phoenix/mapreduce/CsvBulkLoadTool.java | 514 +------------------
.../phoenix/mapreduce/CsvToKeyValueMapper.java | 226 +-------
.../phoenix/mapreduce/CsvToKeyValueReducer.java | 55 --
.../mapreduce/FormatToKeyValueMapper.java | 259 ++++++++++
.../mapreduce/FormatToKeyValueReducer.java | 54 ++
.../ImportPreUpsertKeyValueProcessor.java | 3 +-
.../phoenix/mapreduce/JsonBulkLoadTool.java | 53 ++
.../phoenix/mapreduce/JsonToKeyValueMapper.java | 75 +++
.../mapreduce/MultiHfileOutputFormat.java | 38 +-
.../mapreduce/bulkload/CsvTableRowkeyPair.java | 139 -----
.../mapreduce/bulkload/TableRowkeyPair.java | 134 +++++
.../mapreduce/bulkload/TargetTableRef.java | 70 +++
.../bulkload/TargetTableRefFunctions.java | 95 ++++
.../util/PhoenixConfigurationUtil.java | 15 +-
.../apache/phoenix/util/CSVCommonsLoader.java | 160 +-----
.../org/apache/phoenix/util/SchemaUtil.java | 145 +++++-
.../org/apache/phoenix/util/UpsertExecutor.java | 156 ++++++
.../phoenix/util/csv/CsvUpsertExecutor.java | 131 +----
.../phoenix/util/json/JsonUpsertExecutor.java | 209 ++++++++
.../util/json/ObjectToArrayConverter.java | 69 +++
.../phoenix/mapreduce/BulkLoadToolTest.java | 78 +++
.../mapreduce/CsvBulkImportUtilTest.java | 18 +-
.../phoenix/mapreduce/CsvBulkLoadToolTest.java | 69 ---
.../mapreduce/CsvToKeyValueMapperTest.java | 84 +--
.../mapreduce/FormatToKeyValueMapperTest.java | 102 ++++
.../util/AbstractUpsertExecutorTest.java | 136 +++++
.../phoenix/util/csv/CsvUpsertExecutorTest.java | 144 ++----
.../util/json/JsonUpsertExecutorTest.java | 53 ++
31 files changed, 2222 insertions(+), 1500 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/578979a1/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
index 0e74d7b..a5b7488 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/mapreduce/CsvBulkLoadToolIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/mapreduce/CsvBulkLoadToolIT.java
@@ -17,14 +17,6 @@
*/
package org.apache.phoenix.mapreduce;
-import static org.apache.phoenix.query.BaseTest.setUpConfigForMiniCluster;
-import static org.apache.phoenix.query.QueryServices.DATE_FORMAT_ATTRIB;
-import static org.junit.Assert.assertArrayEquals;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.fail;
-
import java.io.PrintWriter;
import java.sql.Connection;
import java.sql.DriverManager;
@@ -47,6 +39,14 @@ import org.junit.BeforeClass;
import org.junit.Test;
import org.junit.experimental.categories.Category;
+import static org.apache.phoenix.query.BaseTest.setUpConfigForMiniCluster;
+import static org.apache.phoenix.query.QueryServices.DATE_FORMAT_ATTRIB;
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
@Category(NeedsOwnMiniClusterTest.class)
public class CsvBulkLoadToolIT {
http://git-wip-us.apache.org/repos/asf/phoenix/blob/578979a1/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/AbstractBulkLoadTool.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/AbstractBulkLoadTool.java b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/AbstractBulkLoadTool.java
new file mode 100644
index 0000000..cf9ddef
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/AbstractBulkLoadTool.java
@@ -0,0 +1,402 @@
+/*
+ * 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.Preconditions;
+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.HBaseConfiguration;
+import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.client.HTable;
+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.phoenix.jdbc.PhoenixConnection;
+import org.apache.phoenix.jdbc.PhoenixDatabaseMetaData;
+import org.apache.phoenix.jdbc.PhoenixDriver;
+import org.apache.phoenix.mapreduce.bulkload.TableRowkeyPair;
+import org.apache.phoenix.mapreduce.bulkload.TargetTableRef;
+import org.apache.phoenix.mapreduce.bulkload.TargetTableRefFunctions;
+import org.apache.phoenix.query.QueryConstants;
+import org.apache.phoenix.schema.PTable;
+import org.apache.phoenix.util.ColumnInfo;
+import org.apache.phoenix.util.PhoenixRuntime;
+import org.apache.phoenix.util.QueryUtil;
+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.ResultSet;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.ExecutionException;
+
+/**
+ * Base tool for running MapReduce-based ingests of data.
+ */
+public abstract class AbstractBulkLoadTool extends Configured implements Tool {
+
+ protected static final Logger LOG = LoggerFactory.getLogger(AbstractBulkLoadTool.class);
+
+ static final Option ZK_QUORUM_OPT = new Option("z", "zookeeper", true, "Supply zookeeper connection details (optional)");
+ static final Option INPUT_PATH_OPT = new Option("i", "input", true, "Input 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 INDEX_TABLE_NAME_OPT = new Option("it", "index-table", true, "Phoenix index table name when just loading this particualar index table");
+ 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");
+
+ /**
+ * 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
+ */
+ protected abstract void configureOptions(CommandLine cmdLine, List<ColumnInfo> importColumns,
+ Configuration conf) throws SQLException;
+ protected abstract void setupJob(Job job);
+
+ protected Options getOptions() {
+ Options options = new Options();
+ options.addOption(INPUT_PATH_OPT);
+ options.addOption(TABLE_NAME_OPT);
+ options.addOption(INDEX_TABLE_NAME_OPT);
+ options.addOption(ZK_QUORUM_OPT);
+ options.addOption(OUTPUT_PATH_OPT);
+ options.addOption(SCHEMA_NAME_OPT);
+ options.addOption(IMPORT_COLUMNS_OPT);
+ options.addOption(IGNORE_ERRORS_OPT);
+ options.addOption(HELP_OPT);
+ return options;
+ }
+
+ /**
+ * Parses the commandline arguments, throws IllegalStateException if mandatory arguments are
+ * missing.
+ *
+ * @param args supplied command line arguments
+ * @return the parsed command line
+ */
+ protected 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, 0);
+ }
+
+ 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 void printHelpAndExit(String errorMessage, Options options) {
+ System.err.println(errorMessage);
+ printHelpAndExit(options, 1);
+ }
+
+ private void printHelpAndExit(Options options, int exitCode) {
+ HelpFormatter formatter = new HelpFormatter();
+ formatter.printHelp("help", options);
+ System.exit(exitCode);
+ }
+
+ @Override
+ public int run(String[] args) throws Exception {
+
+ Configuration conf = HBaseConfiguration.create(getConf());
+
+ CommandLine cmdLine = null;
+ try {
+ cmdLine = parseOptions(args);
+ } catch (IllegalStateException e) {
+ printHelpAndExit(e.getMessage(), getOptions());
+ }
+ return loadData(conf, cmdLine);
+ }
+
+ private int loadData(Configuration conf, CommandLine cmdLine) throws SQLException,
+ InterruptedException, ExecutionException, ClassNotFoundException {
+ String tableName = cmdLine.getOptionValue(TABLE_NAME_OPT.getOpt());
+ String schemaName = cmdLine.getOptionValue(SCHEMA_NAME_OPT.getOpt());
+ String indexTableName = cmdLine.getOptionValue(INDEX_TABLE_NAME_OPT.getOpt());
+ String qualifiedTableName = getQualifiedTableName(schemaName, tableName);
+ String qualifiedIndexTableName = null;
+ if (indexTableName != null){
+ qualifiedIndexTableName = getQualifiedTableName(schemaName, indexTableName);
+ }
+
+ if (cmdLine.hasOption(ZK_QUORUM_OPT.getOpt())) {
+ // ZK_QUORUM_OPT is optional, but if it's there, use it for both the conn and the job.
+ String zkQuorum = cmdLine.getOptionValue(ZK_QUORUM_OPT.getOpt());
+ PhoenixDriver.ConnectionInfo info = PhoenixDriver.ConnectionInfo.create(zkQuorum);
+ LOG.info("Configuring HBase connection to {}", info);
+ for (Map.Entry<String,String> entry : info.asProps()) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Setting {} = {}", entry.getKey(), entry.getValue());
+ }
+ conf.set(entry.getKey(), entry.getValue());
+ }
+ }
+
+ final Connection conn = QueryUtil.getConnection(conf);
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Reading columns from {} :: {}", ((PhoenixConnection) conn).getURL(),
+ qualifiedTableName);
+ }
+ List<ColumnInfo> importColumns = buildImportColumns(conn, cmdLine, qualifiedTableName);
+ Preconditions.checkNotNull(importColumns);
+ Preconditions.checkArgument(!importColumns.isEmpty(), "Column info list is empty");
+ FormatToKeyValueMapper.configureColumnInfoList(conf, importColumns);
+ boolean ignoreInvalidRows = cmdLine.hasOption(IGNORE_ERRORS_OPT.getOpt());
+ conf.setBoolean(FormatToKeyValueMapper.IGNORE_INVALID_ROW_CONFKEY, ignoreInvalidRows);
+ conf.set(FormatToKeyValueMapper.TABLE_NAME_CONFKEY, tableName);
+
+ // give subclasses their hook
+ configureOptions(cmdLine, importColumns, conf);
+ try {
+ validateTable(conn, schemaName, tableName);
+ } finally {
+ conn.close();
+ }
+
+ final Path inputPath = new Path(cmdLine.getOptionValue(INPUT_PATH_OPT.getOpt()));
+ final Path outputPath;
+ if (cmdLine.hasOption(OUTPUT_PATH_OPT.getOpt())) {
+ outputPath = new Path(cmdLine.getOptionValue(OUTPUT_PATH_OPT.getOpt()));
+ } else {
+ outputPath = new Path("/tmp/" + UUID.randomUUID());
+ }
+
+ List<TargetTableRef> tablesToBeLoaded = new ArrayList<TargetTableRef>();
+ tablesToBeLoaded.add(new TargetTableRef(qualifiedTableName));
+ // using conn after it's been closed... o.O
+ tablesToBeLoaded.addAll(getIndexTables(conn, schemaName, qualifiedTableName));
+
+ // When loading a single index table, check index table name is correct
+ if (qualifiedIndexTableName != null){
+ TargetTableRef targetIndexRef = null;
+ for (TargetTableRef tmpTable : tablesToBeLoaded){
+ if (tmpTable.getLogicalName().compareToIgnoreCase(qualifiedIndexTableName) == 0) {
+ targetIndexRef = tmpTable;
+ break;
+ }
+ }
+ if (targetIndexRef == null){
+ throw new IllegalStateException("Bulk Loader error: index table " +
+ qualifiedIndexTableName + " doesn't exist");
+ }
+ tablesToBeLoaded.clear();
+ tablesToBeLoaded.add(targetIndexRef);
+ }
+
+ return submitJob(conf, tableName, inputPath, outputPath, tablesToBeLoaded);
+ }
+
+ /**
+ * Submits the jobs to the cluster.
+ * Loads the HFiles onto the respective tables.
+ */
+ public int submitJob(final Configuration conf, final String qualifiedTableName, final Path inputPath,
+ final Path outputPath , List<TargetTableRef> tablesToBeLoaded) {
+ try {
+ Job job = new Job(conf, "Phoenix MapReduce import for " + qualifiedTableName);
+ FileInputFormat.addInputPath(job, inputPath);
+ FileOutputFormat.setOutputPath(job, outputPath);
+
+ job.setInputFormatClass(TextInputFormat.class);
+ job.setMapOutputKeyClass(TableRowkeyPair.class);
+ job.setMapOutputValueClass(KeyValue.class);
+ job.setOutputKeyClass(TableRowkeyPair.class);
+ job.setOutputValueClass(KeyValue.class);
+ job.setReducerClass(FormatToKeyValueReducer.class);
+
+ MultiHfileOutputFormat.configureIncrementalLoad(job, tablesToBeLoaded);
+
+ final String tableNamesAsJson = TargetTableRefFunctions.NAMES_TO_JSON.apply(tablesToBeLoaded);
+ job.getConfiguration().set(FormatToKeyValueMapper.TABLE_NAMES_CONFKEY,tableNamesAsJson);
+
+ // give subclasses their hook
+ setupJob(job);
+
+ LOG.info("Running MapReduce import job from {} to {}", inputPath, outputPath);
+ boolean success = job.waitForCompletion(true);
+
+ if (success) {
+ LOG.info("Loading HFiles from {}", outputPath);
+ completebulkload(conf,outputPath,tablesToBeLoaded);
+ }
+
+ LOG.info("Removing output directory {}", outputPath);
+ if (!FileSystem.get(conf).delete(outputPath, true)) {
+ LOG.error("Removing output directory {} failed", outputPath);
+ }
+ return 0;
+ } catch(Exception e) {
+ LOG.error("Error {} occurred submitting BulkLoad ",e.getMessage());
+ return -1;
+ }
+
+ }
+
+ private void completebulkload(Configuration conf,Path outputPath , List<TargetTableRef> tablesToBeLoaded) throws Exception {
+ for(TargetTableRef table : tablesToBeLoaded) {
+ LoadIncrementalHFiles loader = new LoadIncrementalHFiles(conf);
+ String tableName = table.getPhysicalName();
+ Path tableOutputPath = new Path(outputPath,tableName);
+ HTable htable = new HTable(conf,tableName);
+ LOG.info("Loading HFiles for {} from {}", tableName , tableOutputPath);
+ loader.doBulkLoad(tableOutputPath, htable);
+ LOG.info("Incremental load complete for table=" + tableName);
+ }
+ }
+
+ /**
+ * 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 SchemaUtil.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);
+ }
+ }
+
+ /**
+ * 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(
+ "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("Bulk Loader error: Column family name " +
+ "must not start with '_': " + familyName);
+ }
+ }
+ }
+ rs.close();
+ }
+
+ /**
+ * Get the index tables of current data table
+ * @throws java.sql.SQLException
+ */
+ private List<TargetTableRef> getIndexTables(Connection conn, String schemaName, String qualifiedTableName)
+ throws SQLException {
+ PTable table = PhoenixRuntime.getTable(conn, qualifiedTableName);
+ List<TargetTableRef> indexTables = new ArrayList<TargetTableRef>();
+ for(PTable indexTable : table.getIndexes()){
+ if (indexTable.getIndexType() == PTable.IndexType.LOCAL) {
+ throw new UnsupportedOperationException("Local indexes not supported by Bulk Loader");
+ /*indexTables.add(
+ new TargetTableRef(getQualifiedTableName(schemaName,
+ indexTable.getTableName().getString()),
+ MetaDataUtil.getLocalIndexTableName(qualifiedTableName))); */
+ } else {
+ indexTables.add(new TargetTableRef(getQualifiedTableName(schemaName,
+ indexTable.getTableName().getString())));
+ }
+ }
+ return indexTables;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/578979a1/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
index 6d77cd5..bdc67f9 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/CsvBulkImportUtil.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/CsvBulkImportUtil.java
@@ -17,15 +17,11 @@
*/
package org.apache.phoenix.mapreduce;
-import java.util.List;
-
-import com.google.common.annotations.VisibleForTesting;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.util.Base64;
import org.apache.phoenix.mapreduce.util.PhoenixConfigurationUtil;
-import org.apache.phoenix.util.ColumnInfo;
-import com.google.common.base.Preconditions;
+import com.google.common.annotations.VisibleForTesting;
/**
* Collection of utility methods for setting up bulk import jobs.
@@ -36,29 +32,19 @@ 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 quoteChar quote character for the CSV input
* @param escapeChar escape 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, char quoteChar, char escapeChar,
- 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);
+ public static void initCsvImportJob(Configuration conf, char fieldDelimiter, char quoteChar,
+ char escapeChar, String arrayDelimiter) {
setChar(conf, CsvToKeyValueMapper.FIELD_DELIMITER_CONFKEY, fieldDelimiter);
setChar(conf, CsvToKeyValueMapper.QUOTE_CHAR_CONFKEY, quoteChar);
setChar(conf, CsvToKeyValueMapper.ESCAPE_CHAR_CONFKEY, escapeChar);
if (arrayDelimiter != null) {
conf.set(CsvToKeyValueMapper.ARRAY_DELIMITER_CONFKEY, arrayDelimiter);
}
- CsvToKeyValueMapper.configureColumnInfoList(conf, columnInfoList);
- conf.setBoolean(CsvToKeyValueMapper.IGNORE_INVALID_ROW_CONFKEY, ignoreInvalidRows);
}
/**
http://git-wip-us.apache.org/repos/asf/phoenix/blob/578979a1/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
index 20f05ff..e0b083e 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/CsvBulkLoadTool.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/CsvBulkLoadTool.java
@@ -17,364 +17,37 @@
*/
package org.apache.phoenix.mapreduce;
-import java.io.IOException;
-import java.sql.Connection;
-import java.sql.ResultSet;
import java.sql.SQLException;
-import java.util.ArrayList;
import java.util.List;
-import java.util.Map;
-import java.util.UUID;
-import java.util.concurrent.ExecutionException;
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.HBaseConfiguration;
-import org.apache.hadoop.hbase.KeyValue;
-import org.apache.hadoop.hbase.client.HTable;
-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.PhoenixConnection;
-import org.apache.phoenix.jdbc.PhoenixDatabaseMetaData;
-import org.apache.phoenix.jdbc.PhoenixDriver;
-import org.apache.phoenix.mapreduce.bulkload.CsvTableRowkeyPair;
-import org.apache.phoenix.query.QueryConstants;
-import org.apache.phoenix.schema.PTable;
-import org.apache.phoenix.schema.PTable.IndexType;
-import org.apache.phoenix.util.CSVCommonsLoader;
import org.apache.phoenix.util.ColumnInfo;
-import org.apache.phoenix.util.PhoenixRuntime;
-import org.apache.phoenix.util.QueryUtil;
-import org.apache.phoenix.util.SchemaUtil;
-import org.apache.phoenix.util.StringUtil;
-import org.codehaus.jackson.annotate.JsonCreator;
-import org.codehaus.jackson.annotate.JsonProperty;
-import org.codehaus.jackson.map.ObjectMapper;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import com.google.common.annotations.VisibleForTesting;
-import com.google.common.base.Function;
-import com.google.common.base.Splitter;
-import com.google.common.collect.Lists;
-import com.google.common.collect.Maps;
+public class CsvBulkLoadTool extends AbstractBulkLoadTool {
-/**
- * Base tool for running MapReduce-based ingests of data.
- */
-@SuppressWarnings("deprecation")
-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, "Supply zookeeper connection details (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 INDEX_TABLE_NAME_OPT = new Option("it", "index-table", true, "Phoenix index table name when just loading this particualar index table");
static final Option DELIMITER_OPT = new Option("d", "delimiter", true, "Input delimiter, defaults to comma");
static final Option QUOTE_OPT = new Option("q", "quote", true, "Supply a custom phrase delimiter, defaults to double quote character");
static final Option ESCAPE_OPT = new Option("e", "escape", true, "Supply a custom escape character, default is a backslash");
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 {
- int exitStatus = ToolRunner.run(new CsvBulkLoadTool(), args);
- System.exit(exitStatus);
- }
-
- /**
- * Parses the commandline arguments, throws IllegalStateException if mandatory arguments 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, 0);
- }
-
- 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(INDEX_TABLE_NAME_OPT);
- options.addOption(ZK_QUORUM_OPT);
- options.addOption(OUTPUT_PATH_OPT);
- options.addOption(SCHEMA_NAME_OPT);
+ @Override
+ protected Options getOptions() {
+ Options options = super.getOptions();
options.addOption(DELIMITER_OPT);
options.addOption(QUOTE_OPT);
options.addOption(ESCAPE_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, 1);
- }
-
- private void printHelpAndExit(Options options, int exitCode) {
- HelpFormatter formatter = new HelpFormatter();
- formatter.printHelp("help", options);
- System.exit(exitCode);
- }
-
@Override
- public int run(String[] args) throws Exception {
-
- Configuration conf = HBaseConfiguration.create(getConf());
-
- CommandLine cmdLine = null;
- try {
- cmdLine = parseOptions(args);
- } catch (IllegalStateException e) {
- printHelpAndExit(e.getMessage(), getOptions());
- }
- return loadData(conf, cmdLine);
- }
-
- private int loadData(Configuration conf, CommandLine cmdLine) throws SQLException,
- InterruptedException, ExecutionException, ClassNotFoundException {
- String tableName = cmdLine.getOptionValue(TABLE_NAME_OPT.getOpt());
- String schemaName = cmdLine.getOptionValue(SCHEMA_NAME_OPT.getOpt());
- String indexTableName = cmdLine.getOptionValue(INDEX_TABLE_NAME_OPT.getOpt());
- String qualifiedTableName = getQualifiedTableName(schemaName, tableName);
- String qualifiedIndexTableName = null;
- if (indexTableName != null){
- qualifiedIndexTableName = getQualifiedTableName(schemaName, indexTableName);
- }
-
- if (cmdLine.hasOption(ZK_QUORUM_OPT.getOpt())) {
- // ZK_QUORUM_OPT is optional, but if it's there, use it for both the conn and the job.
- String zkQuorum = cmdLine.getOptionValue(ZK_QUORUM_OPT.getOpt());
- PhoenixDriver.ConnectionInfo info = PhoenixDriver.ConnectionInfo.create(zkQuorum);
- LOG.info("Configuring HBase connection to {}", info);
- for (Map.Entry<String,String> entry : info.asProps()) {
- if (LOG.isDebugEnabled()) {
- LOG.debug("Setting {} = {}", entry.getKey(), entry.getValue());
- }
- conf.set(entry.getKey(), entry.getValue());
- }
- }
-
- final Connection conn = QueryUtil.getConnection(conf);
- if (LOG.isDebugEnabled()) {
- LOG.debug("Reading columns from {} :: {}", ((PhoenixConnection) conn).getURL(),
- qualifiedTableName);
- }
- List<ColumnInfo> importColumns = buildImportColumns(conn, cmdLine, qualifiedTableName);
- configureOptions(cmdLine, importColumns, conf);
- try {
- validateTable(conn, schemaName, tableName);
- } finally {
- conn.close();
- }
-
- final Path inputPath = new Path(cmdLine.getOptionValue(INPUT_PATH_OPT.getOpt()));
- final Path outputPath;
- if (cmdLine.hasOption(OUTPUT_PATH_OPT.getOpt())) {
- outputPath = new Path(cmdLine.getOptionValue(OUTPUT_PATH_OPT.getOpt()));
- } else {
- outputPath = new Path("/tmp/" + UUID.randomUUID());
- }
-
- List<TargetTableRef> tablesToBeLoaded = new ArrayList<TargetTableRef>();
- tablesToBeLoaded.add(new TargetTableRef(qualifiedTableName));
- // using conn after it's been closed... o.O
- tablesToBeLoaded.addAll(getIndexTables(conn, schemaName, qualifiedTableName));
-
- // When loading a single index table, check index table name is correct
- if (qualifiedIndexTableName != null){
- TargetTableRef targetIndexRef = null;
- for (TargetTableRef tmpTable : tablesToBeLoaded){
- if (tmpTable.getLogicalName().compareToIgnoreCase(qualifiedIndexTableName) == 0) {
- targetIndexRef = tmpTable;
- break;
- }
- }
- if (targetIndexRef == null){
- throw new IllegalStateException("CSV Bulk Loader error: index table " +
- qualifiedIndexTableName + " doesn't exist");
- }
- tablesToBeLoaded.clear();
- tablesToBeLoaded.add(targetIndexRef);
- }
-
- return submitJob(conf, tableName, inputPath, outputPath, tablesToBeLoaded);
- }
-
- /**
- * Submits the jobs to the cluster.
- * Loads the HFiles onto the respective tables.
- * @param configuration
- * @param qualifiedTableName
- * @param inputPath
- * @param outputPath
- * @param tablesToBeoaded
- * @return status
- */
- public int submitJob(final Configuration conf, final String qualifiedTableName, final Path inputPath,
- final Path outputPath , List<TargetTableRef> tablesToBeLoaded) {
- try {
- Job job = new Job(conf, "Phoenix MapReduce import for " + qualifiedTableName);
-
- // 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);
- FileOutputFormat.setOutputPath(job, outputPath);
- job.setMapperClass(CsvToKeyValueMapper.class);
- job.setMapOutputKeyClass(CsvTableRowkeyPair.class);
- job.setMapOutputValueClass(KeyValue.class);
- job.setOutputKeyClass(CsvTableRowkeyPair.class);
- job.setOutputValueClass(KeyValue.class);
- job.setReducerClass(CsvToKeyValueReducer.class);
-
- MultiHfileOutputFormat.configureIncrementalLoad(job, tablesToBeLoaded);
-
- final String tableNamesAsJson = TargetTableRefFunctions.NAMES_TO_JSON.apply(tablesToBeLoaded);
- job.getConfiguration().set(CsvToKeyValueMapper.TABLE_NAMES_CONFKEY,tableNamesAsJson);
-
- LOG.info("Running MapReduce import job from {} to {}", inputPath, outputPath);
- boolean success = job.waitForCompletion(true);
-
- if (success) {
- LOG.info("Loading HFiles from {}", outputPath);
- completebulkload(conf,outputPath,tablesToBeLoaded);
- }
-
- LOG.info("Removing output directory {}", outputPath);
- if (!FileSystem.get(conf).delete(outputPath, true)) {
- LOG.error("Removing output directory {} failed", outputPath);
- }
- return 0;
- } catch(Exception e) {
- LOG.error("Error {} occurred submitting CSVBulkLoad ",e.getMessage());
- return -1;
- }
-
- }
-
- /**
- * bulkload HFiles .
- * @param conf
- * @param outputPath
- * @param tablesToBeLoaded
- * @throws Exception
- */
- private void completebulkload(Configuration conf,Path outputPath , List<TargetTableRef> tablesToBeLoaded) throws Exception {
- for(TargetTableRef table : tablesToBeLoaded) {
- LoadIncrementalHFiles loader = new LoadIncrementalHFiles(conf);
- String tableName = table.getPhysicalName();
- Path tableOutputPath = new Path(outputPath,tableName);
- HTable htable = new HTable(conf,tableName);
- LOG.info("Loading HFiles for {} from {}", tableName , tableOutputPath);
- loader.doBulkLoad(tableOutputPath, htable);
- LOG.info("Incremental load complete for table=" + tableName);
- }
- }
-
- /**
- * 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
- */
- private static void configureOptions(CommandLine cmdLine, List<ColumnInfo> importColumns,
- Configuration conf) throws SQLException {
+ protected void configureOptions(CommandLine cmdLine, List<ColumnInfo> importColumns,
+ Configuration conf) throws SQLException {
// we don't parse ZK_QUORUM_OPT here because we need it in order to
// create the connection we need to build importColumns.
@@ -408,178 +81,23 @@ public class CsvBulkLoadTool extends Configured implements Tool {
CsvBulkImportUtil.initCsvImportJob(
conf,
- getQualifiedTableName(
- cmdLine.getOptionValue(SCHEMA_NAME_OPT.getOpt()),
- cmdLine.getOptionValue(TABLE_NAME_OPT.getOpt())),
delimiterChar,
quoteChar,
escapeChar,
- cmdLine.getOptionValue(ARRAY_DELIMITER_OPT.getOpt()),
- importColumns,
- cmdLine.hasOption(IGNORE_ERRORS_OPT.getOpt()));
+ cmdLine.getOptionValue(ARRAY_DELIMITER_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();
- }
-
- /**
- * Get the index tables of current data table
- * @throws java.sql.SQLException
- */
- private List<TargetTableRef> getIndexTables(Connection conn, String schemaName, String qualifiedTableName)
- throws SQLException {
- PTable table = PhoenixRuntime.getTable(conn, qualifiedTableName);
- List<TargetTableRef> indexTables = new ArrayList<TargetTableRef>();
- for(PTable indexTable : table.getIndexes()){
- if (indexTable.getIndexType() == IndexType.LOCAL) {
- throw new UnsupportedOperationException("Local indexes not supported by CSV Bulk Loader");
- /*indexTables.add(
- new TargetTableRef(getQualifiedTableName(schemaName,
- indexTable.getTableName().getString()),
- MetaDataUtil.getLocalIndexTableName(qualifiedTableName))); */
- } else {
- indexTables.add(new TargetTableRef(getQualifiedTableName(schemaName,
- indexTable.getTableName().getString())));
- }
+ @Override
+ protected void setupJob(Job job) {
+ // Allow overriding the job jar setting by using a -D system property at startup
+ if (job.getJar() == null) {
+ job.setJarByClass(CsvToKeyValueMapper.class);
}
- return indexTables;
+ job.setMapperClass(CsvToKeyValueMapper.class);
}
- /**
- * Represents the logical and physical name of a single table to which data is to be loaded.
- *
- * This class exists to allow for the difference between HBase physical table names and
- * Phoenix logical table names.
- */
- static class TargetTableRef {
-
- @JsonProperty
- private final String logicalName;
-
- @JsonProperty
- private final String physicalName;
-
- @JsonProperty
- private Map<String,String> configuration = Maps.newHashMap();
-
- private TargetTableRef(String name) {
- this(name, name);
- }
-
- @JsonCreator
- private TargetTableRef(@JsonProperty("logicalName") String logicalName, @JsonProperty("physicalName") String physicalName) {
- this.logicalName = logicalName;
- this.physicalName = physicalName;
- }
-
- public String getLogicalName() {
- return logicalName;
- }
-
- public String getPhysicalName() {
- return physicalName;
- }
-
- public Map<String, String> getConfiguration() {
- return configuration;
- }
-
- public void setConfiguration(Map<String, String> configuration) {
- this.configuration = configuration;
- }
+ public static void main(String[] args) throws Exception {
+ int exitStatus = ToolRunner.run(new CsvBulkLoadTool(), args);
+ System.exit(exitStatus);
}
-
- /**
- * Utility functions to get/put json.
- *
- */
- static class TargetTableRefFunctions {
-
- public static Function<TargetTableRef,String> TO_JSON = new Function<TargetTableRef,String>() {
-
- @Override
- public String apply(TargetTableRef input) {
- try {
- ObjectMapper mapper = new ObjectMapper();
- return mapper.writeValueAsString(input);
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
-
- }
- };
-
- public static Function<String,TargetTableRef> FROM_JSON = new Function<String,TargetTableRef>() {
-
- @Override
- public TargetTableRef apply(String json) {
- try {
- ObjectMapper mapper = new ObjectMapper();
- return mapper.readValue(json, TargetTableRef.class);
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
-
- }
- };
-
- public static Function<List<TargetTableRef>,String> NAMES_TO_JSON = new Function<List<TargetTableRef>,String>() {
-
- @Override
- public String apply(List<TargetTableRef> input) {
- try {
- List<String> tableNames = Lists.newArrayListWithCapacity(input.size());
- for(TargetTableRef table : input) {
- tableNames.add(table.getPhysicalName());
- }
- ObjectMapper mapper = new ObjectMapper();
- return mapper.writeValueAsString(tableNames);
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
-
- }
- };
-
- public static Function<String,List<String>> NAMES_FROM_JSON = new Function<String,List<String>>() {
-
- @SuppressWarnings("unchecked")
- @Override
- public List<String> apply(String json) {
- try {
- ObjectMapper mapper = new ObjectMapper();
- return mapper.readValue(json, ArrayList.class);
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
-
- }
- };
- }
-
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/578979a1/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
index c3b5a7d..5a5d378 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/CsvToKeyValueMapper.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/CsvToKeyValueMapper.java
@@ -19,46 +19,20 @@ package org.apache.phoenix.mapreduce;
import java.io.IOException;
import java.io.StringReader;
-import java.sql.SQLException;
-import java.util.Iterator;
import java.util.List;
-import java.util.Map.Entry;
-import java.util.Properties;
-
-import javax.annotation.Nullable;
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.KeyValue;
-import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
-import org.apache.hadoop.hbase.util.Bytes;
-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.jdbc.PhoenixConnection;
-import org.apache.phoenix.mapreduce.CsvBulkLoadTool.TargetTableRefFunctions;
-import org.apache.phoenix.mapreduce.bulkload.CsvTableRowkeyPair;
-import org.apache.phoenix.mapreduce.util.PhoenixConfigurationUtil;
import org.apache.phoenix.util.CSVCommonsLoader;
import org.apache.phoenix.util.ColumnInfo;
-import org.apache.phoenix.util.PhoenixRuntime;
-import org.apache.phoenix.util.QueryUtil;
+import org.apache.phoenix.util.UpsertExecutor;
import org.apache.phoenix.util.csv.CsvUpsertExecutor;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
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.base.Throwables;
-import com.google.common.collect.ImmutableList;
import com.google.common.collect.Iterables;
-import com.google.common.collect.Lists;
/**
* MapReduce mapper that converts CSV input lines into KeyValues that can be written to HFiles.
@@ -67,12 +41,7 @@ import com.google.common.collect.Lists;
* extracting the created KeyValues and rolling back the statement execution before it is
* committed to HBase.
*/
-public class CsvToKeyValueMapper extends Mapper<LongWritable,Text,CsvTableRowkeyPair,
- KeyValue> {
-
- private static final Logger LOG = LoggerFactory.getLogger(CsvToKeyValueMapper.class);
-
- private static final String COUNTER_GROUP_NAME = "Phoenix MapReduce Import";
+public class CsvToKeyValueMapper extends FormatToKeyValueMapper<CSVRecord> {
/** Configuration key for the field delimiter for input csv records */
public static final String FIELD_DELIMITER_CONFKEY = "phoenix.mapreduce.import.fielddelimiter";
@@ -86,117 +55,25 @@ public class CsvToKeyValueMapper extends Mapper<LongWritable,Text,CsvTableRowkey
/** 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 name of the output index table */
- public static final String INDEX_TABLE_NAME_CONFKEY = "phoenix.mapreduce.import.indextablename";
-
- /** Configuration key for the columns to be imported */
- public static final String COLUMN_INFO_CONFKEY = "phoenix.mapreduce.import.columninfos";
+ private CsvLineParser lineParser;
- /** Configuration key for the flag to ignore invalid rows */
- public static final String IGNORE_INVALID_ROW_CONFKEY = "phoenix.mapreduce.import.ignoreinvalidrow";
-
- /** Configuration key for the table names */
- public static final String TABLE_NAMES_CONFKEY = "phoenix.mapreduce.import.tablenames";
-
- /** Configuration key for the table configurations */
- public static final String TABLE_CONFIG_CONFKEY = "phoenix.mapreduce.import.table.config";
-
- private PhoenixConnection conn;
- private CsvUpsertExecutor csvUpsertExecutor;
- private MapperUpsertListener upsertListener;
- private CsvLineParser csvLineParser;
- private ImportPreUpsertKeyValueProcessor preUpdateProcessor;
- private List<String> tableNames;
+ @Override
+ protected LineParser<CSVRecord> getLineParser() {
+ return lineParser;
+ }
@Override
protected void setup(Context context) throws IOException, InterruptedException {
-
Configuration conf = context.getConfiguration();
-
- // pass client configuration into driver
- Properties clientInfos = new Properties();
- Iterator<Entry<String, String>> iterator = conf.iterator();
- while(iterator.hasNext()) {
- Entry<String,String> entry = iterator.next();
- clientInfos.setProperty(entry.getKey(), entry.getValue());
- }
-
- try {
- conn = (PhoenixConnection) QueryUtil.getConnection(clientInfos, conf);
- } catch (SQLException | ClassNotFoundException e) {
- throw new RuntimeException(e);
- }
-
- final String tableNamesConf = conf.get(TABLE_NAMES_CONFKEY);
- tableNames = TargetTableRefFunctions.NAMES_FROM_JSON.apply(tableNamesConf);
-
- upsertListener = new MapperUpsertListener(
- context, conf.getBoolean(IGNORE_INVALID_ROW_CONFKEY, true));
- csvUpsertExecutor = buildUpsertExecutor(conf);
- csvLineParser = new CsvLineParser(
+ lineParser = new CsvLineParser(
CsvBulkImportUtil.getCharacter(conf, FIELD_DELIMITER_CONFKEY),
CsvBulkImportUtil.getCharacter(conf, QUOTE_CHAR_CONFKEY),
CsvBulkImportUtil.getCharacter(conf, ESCAPE_CHAR_CONFKEY));
-
- preUpdateProcessor = PhoenixConfigurationUtil.loadPreUpsertProcessor(conf);
- }
-
- @SuppressWarnings("deprecation")
- @Override
- protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
- 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, true);
- while (uncommittedDataIterator.hasNext()) {
- Pair<byte[], List<KeyValue>> kvPair = uncommittedDataIterator.next();
- List<KeyValue> keyValueList = kvPair.getSecond();
- keyValueList = preUpdateProcessor.preUpsert(kvPair.getFirst(), keyValueList);
- byte[] first = kvPair.getFirst();
- for(String tableName : tableNames) {
- if (Bytes.compareTo(Bytes.toBytes(tableName), first) != 0) {
- // skip edits for other tables
- continue;
- }
- for (KeyValue kv : keyValueList) {
- ImmutableBytesWritable outputKey = new ImmutableBytesWritable();
- outputKey.set(kv.getRowArray(), kv.getRowOffset(), kv.getRowLength());
- context.write(new CsvTableRowkeyPair(tableName, 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);
- }
}
@VisibleForTesting
- CsvUpsertExecutor buildUpsertExecutor(Configuration conf) {
+ @Override
+ protected UpsertExecutor<CSVRecord, ?> buildUpsertExecutor(Configuration conf) {
String tableName = conf.get(TABLE_NAME_CONFKEY);
String arraySeparator = conf.get(ARRAY_DELIMITER_CONFKEY,
CSVCommonsLoader.DEFAULT_ARRAY_ELEMENT_SEPARATOR);
@@ -204,78 +81,14 @@ public class CsvToKeyValueMapper extends Mapper<LongWritable,Text,CsvTableRowkey
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, Throwable throwable) {
- LOG.error("Error on record " + csvRecord, throwable);
- context.getCounter(COUNTER_GROUP_NAME, "Errors on records").increment(1L);
- if (!ignoreRecordErrors) {
- throw Throwables.propagate(throwable);
- }
- }
+ return new CsvUpsertExecutor(conn, tableName, columnInfoList, upsertListener, arraySeparator);
}
/**
* Parses a single CSV input line, returning a {@code CSVRecord}.
*/
@VisibleForTesting
- static class CsvLineParser {
+ static class CsvLineParser implements LineParser<CSVRecord> {
private final CSVFormat csvFormat;
CsvLineParser(char fieldDelimiter, char quote, char escape) {
@@ -286,6 +99,7 @@ public class CsvToKeyValueMapper extends Mapper<LongWritable,Text,CsvTableRowkey
.withQuote(quote);
}
+ @Override
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
@@ -294,18 +108,4 @@ public class CsvToKeyValueMapper extends Mapper<LongWritable,Text,CsvTableRowkey
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/phoenix/blob/578979a1/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/CsvToKeyValueReducer.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/CsvToKeyValueReducer.java b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/CsvToKeyValueReducer.java
deleted file mode 100644
index 7e9c4fd..0000000
--- a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/CsvToKeyValueReducer.java
+++ /dev/null
@@ -1,55 +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.mapreduce;
-
-import java.io.IOException;
-import java.util.TreeSet;
-
-import org.apache.hadoop.hbase.KeyValue;
-import org.apache.hadoop.hbase.mapreduce.KeyValueSortReducer;
-import org.apache.hadoop.mapreduce.Reducer;
-import org.apache.phoenix.mapreduce.bulkload.CsvTableRowkeyPair;
-
-/**
- * Reducer class for the CSVBulkLoad job.
- * Performs similar functionality to {@link KeyValueSortReducer}
- *
- */
-public class CsvToKeyValueReducer extends Reducer<CsvTableRowkeyPair,KeyValue,CsvTableRowkeyPair,KeyValue> {
-
- @Override
- protected void reduce(CsvTableRowkeyPair key, Iterable<KeyValue> values,
- Reducer<CsvTableRowkeyPair, KeyValue, CsvTableRowkeyPair, KeyValue>.Context context)
- throws IOException, InterruptedException {
- TreeSet<KeyValue> map = new TreeSet<KeyValue>(KeyValue.COMPARATOR);
- for (KeyValue kv: values) {
- try {
- map.add(kv.clone());
- } catch (CloneNotSupportedException e) {
- throw new java.io.IOException(e);
- }
- }
- context.setStatus("Read " + map.getClass());
- int index = 0;
- for (KeyValue kv: map) {
- context.write(key, kv);
- if (++index % 100 == 0) context.setStatus("Wrote " + index);
- }
- }
-
-}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/578979a1/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/FormatToKeyValueMapper.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/FormatToKeyValueMapper.java b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/FormatToKeyValueMapper.java
new file mode 100644
index 0000000..b2e99e5
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/FormatToKeyValueMapper.java
@@ -0,0 +1,259 @@
+/*
+ * 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 java.io.IOException;
+import java.sql.SQLException;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import javax.annotation.Nullable;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
+import org.apache.hadoop.hbase.util.Bytes;
+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.jdbc.PhoenixConnection;
+import org.apache.phoenix.mapreduce.bulkload.TableRowkeyPair;
+import org.apache.phoenix.mapreduce.bulkload.TargetTableRefFunctions;
+import org.apache.phoenix.mapreduce.util.PhoenixConfigurationUtil;
+import org.apache.phoenix.util.ColumnInfo;
+import org.apache.phoenix.util.PhoenixRuntime;
+import org.apache.phoenix.util.QueryUtil;
+import org.apache.phoenix.util.UpsertExecutor;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Function;
+import com.google.common.base.Joiner;
+import com.google.common.base.Splitter;
+import com.google.common.base.Throwables;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Iterables;
+import com.google.common.collect.Lists;
+
+/**
+ * Base class for converting some input source format into {@link KeyValue}s of a target
+ * schema. Assumes input format is text-based, with one row per line. Depends on an online cluster
+ * to retrieve {@link ColumnInfo} from the target table.
+ */
+public abstract class FormatToKeyValueMapper<RECORD> extends Mapper<LongWritable, Text, TableRowkeyPair,
+ KeyValue> {
+
+ protected static final Logger LOG = LoggerFactory.getLogger(FormatToKeyValueMapper.class);
+
+ protected static final String COUNTER_GROUP_NAME = "Phoenix MapReduce Import";
+
+ /** Configuration key for the name of the output table */
+ public static final String TABLE_NAME_CONFKEY = "phoenix.mapreduce.import.tablename";
+
+ /** Configuration key for the name of the output index table */
+ public static final String INDEX_TABLE_NAME_CONFKEY = "phoenix.mapreduce.import.indextablename";
+
+ /** 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";
+
+ /** Configuration key for the table names */
+ public static final String TABLE_NAMES_CONFKEY = "phoenix.mapreduce.import.tablenames";
+
+ /** Configuration key for the table configurations */
+ public static final String TABLE_CONFIG_CONFKEY = "phoenix.mapreduce.import.table.config";
+
+ /**
+ * Parses a single input line, returning a {@code T}.
+ */
+ public interface LineParser<T> {
+ T parse(String input) throws IOException;
+ }
+
+ protected PhoenixConnection conn;
+ protected UpsertExecutor<RECORD, ?> upsertExecutor;
+ protected ImportPreUpsertKeyValueProcessor preUpdateProcessor;
+ protected List<String> tableNames;
+ protected MapperUpsertListener<RECORD> upsertListener;
+
+ protected abstract UpsertExecutor<RECORD,?> buildUpsertExecutor(Configuration conf);
+ protected abstract LineParser<RECORD> getLineParser();
+
+ @Override
+ protected void setup(Context context) throws IOException, InterruptedException {
+
+ Configuration conf = context.getConfiguration();
+
+ // pass client configuration into driver
+ Properties clientInfos = new Properties();
+ for (Map.Entry<String, String> entry : conf) {
+ clientInfos.setProperty(entry.getKey(), entry.getValue());
+ }
+
+ try {
+ conn = (PhoenixConnection) QueryUtil.getConnection(clientInfos, conf);
+ } catch (SQLException | ClassNotFoundException e) {
+ throw new RuntimeException(e);
+ }
+
+ final String tableNamesConf = conf.get(TABLE_NAMES_CONFKEY);
+ tableNames = TargetTableRefFunctions.NAMES_FROM_JSON.apply(tableNamesConf);
+
+ upsertListener = new MapperUpsertListener<RECORD>(
+ context, conf.getBoolean(IGNORE_INVALID_ROW_CONFKEY, true));
+ upsertExecutor = buildUpsertExecutor(conf);
+ preUpdateProcessor = PhoenixConfigurationUtil.loadPreUpsertProcessor(conf);
+ }
+
+ @SuppressWarnings("deprecation")
+ @Override
+ protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
+ try {
+ RECORD record = null;
+ try {
+ record = getLineParser().parse(value.toString());
+ } catch (IOException e) {
+ context.getCounter(COUNTER_GROUP_NAME, "Parser errors").increment(1L);
+ return;
+ }
+
+ if (record == null) {
+ context.getCounter(COUNTER_GROUP_NAME, "Empty records").increment(1L);
+ return;
+ }
+ upsertExecutor.execute(ImmutableList.<RECORD>of(record));
+
+ Iterator<Pair<byte[], List<KeyValue>>> uncommittedDataIterator
+ = PhoenixRuntime.getUncommittedDataIterator(conn, true);
+ while (uncommittedDataIterator.hasNext()) {
+ Pair<byte[], List<KeyValue>> kvPair = uncommittedDataIterator.next();
+ List<KeyValue> keyValueList = kvPair.getSecond();
+ keyValueList = preUpdateProcessor.preUpsert(kvPair.getFirst(), keyValueList);
+ byte[] first = kvPair.getFirst();
+ for (String tableName : tableNames) {
+ if (Bytes.compareTo(Bytes.toBytes(tableName), first) != 0) {
+ // skip edits for other tables
+ continue;
+ }
+ for (KeyValue kv : keyValueList) {
+ ImmutableBytesWritable outputKey = new ImmutableBytesWritable();
+ outputKey.set(kv.getRowArray(), kv.getRowOffset(), kv.getRowLength());
+ context.write(new TableRowkeyPair(tableName, 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);
+ }
+ }
+
+ /**
+ * 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 == null || 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<T> implements UpsertExecutor.UpsertListener<T> {
+
+ private final Mapper<LongWritable, Text, TableRowkeyPair, KeyValue>.Context context;
+ private final boolean ignoreRecordErrors;
+
+ private MapperUpsertListener(
+ Mapper<LongWritable, Text, TableRowkeyPair, KeyValue>.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(T record, Throwable throwable) {
+ LOG.error("Error on record " + record, throwable);
+ context.getCounter(COUNTER_GROUP_NAME, "Errors on records").increment(1L);
+ if (!ignoreRecordErrors) {
+ throw Throwables.propagate(throwable);
+ }
+ }
+ }
+
+ /**
+ * 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/phoenix/blob/578979a1/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/FormatToKeyValueReducer.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/FormatToKeyValueReducer.java b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/FormatToKeyValueReducer.java
new file mode 100644
index 0000000..5d00656
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/FormatToKeyValueReducer.java
@@ -0,0 +1,54 @@
+/*
+ * 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 java.io.IOException;
+import java.util.TreeSet;
+
+import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.mapreduce.KeyValueSortReducer;
+import org.apache.hadoop.mapreduce.Reducer;
+import org.apache.phoenix.mapreduce.bulkload.TableRowkeyPair;
+
+/**
+ * Reducer class for the bulkload jobs.
+ * Performs similar functionality to {@link KeyValueSortReducer}
+ */
+public class FormatToKeyValueReducer
+ extends Reducer<TableRowkeyPair,KeyValue,TableRowkeyPair,KeyValue> {
+
+ @Override
+ protected void reduce(TableRowkeyPair key, Iterable<KeyValue> values,
+ Reducer<TableRowkeyPair, KeyValue, TableRowkeyPair, KeyValue>.Context context)
+ throws IOException, InterruptedException {
+ TreeSet<KeyValue> map = new TreeSet<KeyValue>(KeyValue.COMPARATOR);
+ for (KeyValue kv: values) {
+ try {
+ map.add(kv.clone());
+ } catch (CloneNotSupportedException e) {
+ throw new java.io.IOException(e);
+ }
+ }
+ context.setStatus("Read " + map.getClass());
+ int index = 0;
+ for (KeyValue kv: map) {
+ context.write(key, kv);
+ if (++index % 100 == 0) context.setStatus("Wrote " + index);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/578979a1/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
index 62211f3..dff9ef2 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/ImportPreUpsertKeyValueProcessor.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/ImportPreUpsertKeyValueProcessor.java
@@ -24,7 +24,8 @@ 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.
+ * org.apache.phoenix.mapreduce.util.PhoenixConfigurationUtil#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.
http://git-wip-us.apache.org/repos/asf/phoenix/blob/578979a1/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/JsonBulkLoadTool.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/JsonBulkLoadTool.java b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/JsonBulkLoadTool.java
new file mode 100644
index 0000000..1bea3f0
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/JsonBulkLoadTool.java
@@ -0,0 +1,53 @@
+/*
+ * 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 java.sql.SQLException;
+import java.util.List;
+
+import org.apache.commons.cli.CommandLine;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.mapreduce.Job;
+import org.apache.hadoop.util.ToolRunner;
+import org.apache.phoenix.util.ColumnInfo;
+
+/**
+ * A tool for running MapReduce-based ingests of JSON data. Nested JSON data structures are not
+ * handled, though lists are converted into typed ARRAYS.
+ */
+public class JsonBulkLoadTool extends AbstractBulkLoadTool {
+
+ @Override
+ protected void configureOptions(CommandLine cmdLine, List<ColumnInfo> importColumns,
+ Configuration conf) throws SQLException {
+ // noop
+ }
+
+ @Override
+ protected void setupJob(Job job) {
+ // Allow overriding the job jar setting by using a -D system property at startup
+ if (job.getJar() == null) {
+ job.setJarByClass(JsonToKeyValueMapper.class);
+ }
+ job.setMapperClass(JsonToKeyValueMapper.class);
+ }
+
+ public static void main(String[] args) throws Exception {
+ ToolRunner.run(new JsonBulkLoadTool(), args);
+ }
+}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/578979a1/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/JsonToKeyValueMapper.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/JsonToKeyValueMapper.java b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/JsonToKeyValueMapper.java
new file mode 100644
index 0000000..5173a0e
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/JsonToKeyValueMapper.java
@@ -0,0 +1,75 @@
+/*
+ * 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 java.io.IOException;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.phoenix.util.ColumnInfo;
+import org.apache.phoenix.util.UpsertExecutor;
+import org.apache.phoenix.util.json.JsonUpsertExecutor;
+import org.codehaus.jackson.map.ObjectMapper;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+
+/**
+ * MapReduce mapper that converts JSON 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 JsonToKeyValueMapper extends FormatToKeyValueMapper<Map<?, ?>> {
+
+ private LineParser<Map<?, ?>> lineParser;
+
+ @Override
+ protected LineParser<Map<?, ?>> getLineParser() {
+ return lineParser;
+ }
+
+ @Override
+ protected void setup(Context context) throws IOException, InterruptedException {
+ super.setup(context);
+ lineParser = new JsonLineParser();
+ }
+
+ @VisibleForTesting
+ @Override
+ protected UpsertExecutor<Map<?, ?>, ?> buildUpsertExecutor(Configuration conf) {
+ String tableName = conf.get(TABLE_NAME_CONFKEY);
+ Preconditions.checkNotNull(tableName, "table name is not configured");
+
+ List<ColumnInfo> columnInfoList = buildColumnInfoList(conf);
+
+ return new JsonUpsertExecutor(conn, tableName, columnInfoList, upsertListener);
+ }
+
+ @VisibleForTesting
+ static class JsonLineParser implements LineParser<Map<?, ?>> {
+ private final ObjectMapper mapper = new ObjectMapper();
+
+ @Override
+ public Map<?, ?> parse(String input) throws IOException {
+ return mapper.readValue(input, Map.class);
+ }
+ }
+}
[2/3] phoenix git commit: PHOENIX-2481 JSON bulkload tool
Posted by nd...@apache.org.
http://git-wip-us.apache.org/repos/asf/phoenix/blob/578979a1/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/MultiHfileOutputFormat.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/MultiHfileOutputFormat.java b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/MultiHfileOutputFormat.java
index eae58ad..05fbab2 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/MultiHfileOutputFormat.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/MultiHfileOutputFormat.java
@@ -66,9 +66,9 @@ import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.output.FileOutputCommitter;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.partition.TotalOrderPartitioner;
-import org.apache.phoenix.mapreduce.CsvBulkLoadTool.TargetTableRef;
-import org.apache.phoenix.mapreduce.CsvBulkLoadTool.TargetTableRefFunctions;
-import org.apache.phoenix.mapreduce.bulkload.CsvTableRowkeyPair;
+import org.apache.phoenix.mapreduce.bulkload.TableRowkeyPair;
+import org.apache.phoenix.mapreduce.bulkload.TargetTableRef;
+import org.apache.phoenix.mapreduce.bulkload.TargetTableRefFunctions;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -81,7 +81,7 @@ import com.google.common.collect.Sets;
* It has been adapted from {#link HFileOutputFormat2} but differs from the fact it creates
* HFiles for multiple tables.
*/
-public class MultiHfileOutputFormat extends FileOutputFormat<CsvTableRowkeyPair, Cell> {
+public class MultiHfileOutputFormat extends FileOutputFormat<TableRowkeyPair, Cell> {
private static final Logger LOG = LoggerFactory.getLogger(MultiHfileOutputFormat.class);
@@ -101,7 +101,7 @@ public class MultiHfileOutputFormat extends FileOutputFormat<CsvTableRowkeyPair,
private static final String AT_DELIMITER = "@";
@Override
- public RecordWriter<CsvTableRowkeyPair, Cell> getRecordWriter(TaskAttemptContext context)
+ public RecordWriter<TableRowkeyPair, Cell> getRecordWriter(TaskAttemptContext context)
throws IOException, InterruptedException {
return createRecordWriter(context);
}
@@ -112,7 +112,7 @@ public class MultiHfileOutputFormat extends FileOutputFormat<CsvTableRowkeyPair,
* @return
* @throws IOException
*/
- static <V extends Cell> RecordWriter<CsvTableRowkeyPair, V> createRecordWriter(final TaskAttemptContext context)
+ static <V extends Cell> RecordWriter<TableRowkeyPair, V> createRecordWriter(final TaskAttemptContext context)
throws IOException {
// Get the path of the temporary output file
final Path outputPath = FileOutputFormat.getOutputPath(context);
@@ -130,7 +130,7 @@ public class MultiHfileOutputFormat extends FileOutputFormat<CsvTableRowkeyPair,
final boolean compactionExclude = conf.getBoolean(
"hbase.mapreduce.hfileoutputformat.compaction.exclude", false);
- return new RecordWriter<CsvTableRowkeyPair, V>() {
+ return new RecordWriter<TableRowkeyPair, V>() {
// Map of families to writers and how much has been output on the writer.
private final Map<byte [], WriterLength> writers =
new TreeMap<byte [], WriterLength>(Bytes.BYTES_COMPARATOR);
@@ -139,7 +139,7 @@ public class MultiHfileOutputFormat extends FileOutputFormat<CsvTableRowkeyPair,
private boolean rollRequested = false;
@Override
- public void write(CsvTableRowkeyPair row, V cell)
+ public void write(TableRowkeyPair row, V cell)
throws IOException {
KeyValue kv = KeyValueUtil.ensureKeyValue(cell);
// null input == user explicitly wants to flush
@@ -450,7 +450,7 @@ public class MultiHfileOutputFormat extends FileOutputFormat<CsvTableRowkeyPair,
* Configure <code>job</code> with a TotalOrderPartitioner, partitioning against
* <code>splitPoints</code>. Cleans up the partitions file after job exists.
*/
- static void configurePartitioner(Job job, Set<CsvTableRowkeyPair> tablesStartKeys)
+ static void configurePartitioner(Job job, Set<TableRowkeyPair> tablesStartKeys)
throws IOException {
Configuration conf = job.getConfiguration();
@@ -467,7 +467,7 @@ public class MultiHfileOutputFormat extends FileOutputFormat<CsvTableRowkeyPair,
}
private static void writePartitions(Configuration conf, Path partitionsPath,
- Set<CsvTableRowkeyPair> tablesStartKeys) throws IOException {
+ Set<TableRowkeyPair> tablesStartKeys) throws IOException {
LOG.info("Writing partition information to " + partitionsPath);
if (tablesStartKeys.isEmpty()) {
@@ -478,9 +478,9 @@ public class MultiHfileOutputFormat extends FileOutputFormat<CsvTableRowkeyPair,
// have keys < the first region (which has an empty start key)
// so we need to remove it. Otherwise we would end up with an
// empty reducer with index 0
- TreeSet<CsvTableRowkeyPair> sorted = new TreeSet<CsvTableRowkeyPair>(tablesStartKeys);
+ TreeSet<TableRowkeyPair> sorted = new TreeSet<TableRowkeyPair>(tablesStartKeys);
- CsvTableRowkeyPair first = sorted.first();
+ TableRowkeyPair first = sorted.first();
if (!first.getRowkey().equals(HConstants.EMPTY_BYTE_ARRAY)) {
throw new IllegalArgumentException(
"First region of table should have empty start key. Instead has: "
@@ -491,11 +491,11 @@ public class MultiHfileOutputFormat extends FileOutputFormat<CsvTableRowkeyPair,
// Write the actual file
FileSystem fs = partitionsPath.getFileSystem(conf);
SequenceFile.Writer writer = SequenceFile.createWriter(
- fs, conf, partitionsPath, CsvTableRowkeyPair.class,
+ fs, conf, partitionsPath, TableRowkeyPair.class,
NullWritable.class);
try {
- for (CsvTableRowkeyPair startKey : sorted) {
+ for (TableRowkeyPair startKey : sorted) {
writer.append(startKey, NullWritable.get());
}
} finally {
@@ -658,11 +658,11 @@ public class MultiHfileOutputFormat extends FileOutputFormat<CsvTableRowkeyPair,
KeyValueSerialization.class.getName());
// tableStartKeys for all tables.
- Set<CsvTableRowkeyPair> tablesStartKeys = Sets.newTreeSet();
+ Set<TableRowkeyPair> tablesStartKeys = Sets.newTreeSet();
for(TargetTableRef table : tablesToBeLoaded) {
final String tableName = table.getPhysicalName();
try(HTable htable = new HTable(conf,tableName);){
- Set<CsvTableRowkeyPair> startKeys = getRegionStartKeys(tableName , htable.getRegionLocator());
+ Set<TableRowkeyPair> startKeys = getRegionStartKeys(tableName , htable.getRegionLocator());
tablesStartKeys.addAll(startKeys);
String compressionConfig = configureCompression(htable.getTableDescriptor());
String bloomTypeConfig = configureBloomType(htable.getTableDescriptor());
@@ -704,12 +704,12 @@ public class MultiHfileOutputFormat extends FileOutputFormat<CsvTableRowkeyPair,
* Return the start keys of all of the regions in this table,
* as a list of ImmutableBytesWritable.
*/
- private static Set<CsvTableRowkeyPair> getRegionStartKeys(String tableName , RegionLocator table) throws IOException {
+ private static Set<TableRowkeyPair> getRegionStartKeys(String tableName , RegionLocator table) throws IOException {
byte[][] byteKeys = table.getStartKeys();
- Set<CsvTableRowkeyPair> ret = new TreeSet<CsvTableRowkeyPair>();
+ Set<TableRowkeyPair> ret = new TreeSet<TableRowkeyPair>();
for (byte[] byteKey : byteKeys) {
// phoenix-2216: start : passing the table name and startkey
- ret.add(new CsvTableRowkeyPair(tableName, new ImmutableBytesWritable(byteKey)));
+ ret.add(new TableRowkeyPair(tableName, new ImmutableBytesWritable(byteKey)));
}
return ret;
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/578979a1/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/bulkload/CsvTableRowkeyPair.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/bulkload/CsvTableRowkeyPair.java b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/bulkload/CsvTableRowkeyPair.java
deleted file mode 100644
index 3ae74b6..0000000
--- a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/bulkload/CsvTableRowkeyPair.java
+++ /dev/null
@@ -1,139 +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.mapreduce.bulkload;
-
-import java.io.DataInput;
-import java.io.DataOutput;
-import java.io.IOException;
-
-import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
-import org.apache.hadoop.io.BytesWritable;
-import org.apache.hadoop.io.WritableComparable;
-import org.apache.hadoop.io.WritableComparator;
-import org.apache.hadoop.io.WritableUtils;
-
-import com.google.common.base.Preconditions;
-
-
-/**
- * A WritableComparable to hold the table name and the rowkey.
- *
- */
-public class CsvTableRowkeyPair implements WritableComparable<CsvTableRowkeyPair> {
-
- /* The qualified table name */
- private String tableName;
-
- /* The rowkey for the record */
- private ImmutableBytesWritable rowkey;
-
- /**
- * Default constructor
- */
- public CsvTableRowkeyPair() {
- super();
- }
-
- /**
- * @param tableName
- * @param rowkey
- */
- public CsvTableRowkeyPair(String tableName, ImmutableBytesWritable rowkey) {
- super();
- Preconditions.checkNotNull(tableName);
- Preconditions.checkNotNull(rowkey);
- this.tableName = tableName;
- this.rowkey = rowkey;
- }
-
- public String getTableName() {
- return tableName;
- }
-
- public void setTableName(String tableName) {
- this.tableName = tableName;
- }
-
- public ImmutableBytesWritable getRowkey() {
- return rowkey;
- }
-
- public void setRowkey(ImmutableBytesWritable rowkey) {
- this.rowkey = rowkey;
- }
-
- @Override
- public void readFields(DataInput input) throws IOException {
- tableName = WritableUtils.readString(input);
- rowkey = new ImmutableBytesWritable();
- rowkey.readFields(input);
- }
-
- @Override
- public void write(DataOutput output) throws IOException {
- WritableUtils.writeString(output,tableName);
- rowkey.write(output);
- }
-
- @Override
- public int compareTo(CsvTableRowkeyPair other) {
- String otherTableName = other.getTableName();
- if(this.tableName.equals(otherTableName)) {
- return this.rowkey.compareTo(other.getRowkey());
- } else {
- return this.tableName.compareTo(otherTableName);
- }
- }
-
- /** Comparator optimized for <code>CsvTableRowkeyPair</code>. */
- public static class Comparator extends WritableComparator {
- private BytesWritable.Comparator comparator = new BytesWritable.Comparator();
-
- public Comparator() {
- super(CsvTableRowkeyPair.class);
- }
-
- @Override
- public int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2) {
- try {
- int vintL1 = WritableUtils.decodeVIntSize(b1[s1]);
- int vintL2 = WritableUtils.decodeVIntSize(b2[s2]);
- int strL1 = readVInt(b1, s1);
- int strL2 = readVInt(b2, s2);
- int cmp = compareBytes(b1, s1 + vintL1, strL1, b2, s2 + vintL2, strL2);
- if (cmp != 0) {
- return cmp;
- }
- int vintL3 = WritableUtils.decodeVIntSize(b1[s1 + vintL1 + strL1]);
- int vintL4 = WritableUtils.decodeVIntSize(b2[s2 + vintL2 + strL2]);
- int strL3 = readVInt(b1, s1 + vintL1 + strL1);
- int strL4 = readVInt(b2, s2 + vintL2 + strL2);
- return comparator.compare(b1, s1 + vintL1 + strL1 + vintL3, strL3, b2, s2
- + vintL2 + strL2 + vintL4, strL4);
-
- } catch(Exception ex) {
- throw new IllegalArgumentException(ex);
- }
- }
- }
-
- static {
- WritableComparator.define(CsvTableRowkeyPair.class, new Comparator());
- }
-
-}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/578979a1/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/bulkload/TableRowkeyPair.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/bulkload/TableRowkeyPair.java b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/bulkload/TableRowkeyPair.java
new file mode 100644
index 0000000..412226f
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/bulkload/TableRowkeyPair.java
@@ -0,0 +1,134 @@
+/*
+ * 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.bulkload;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+
+import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
+import org.apache.hadoop.io.BytesWritable;
+import org.apache.hadoop.io.WritableComparable;
+import org.apache.hadoop.io.WritableComparator;
+import org.apache.hadoop.io.WritableUtils;
+
+import com.google.common.base.Preconditions;
+
+
+/**
+ * A WritableComparable to hold the table name and the rowkey.
+ */
+public class TableRowkeyPair implements WritableComparable<TableRowkeyPair> {
+
+ /* The qualified table name */
+ private String tableName;
+
+ /* The rowkey for the record */
+ private ImmutableBytesWritable rowkey;
+
+ /**
+ * Default constructor
+ */
+ public TableRowkeyPair() {
+ super();
+ }
+
+ public TableRowkeyPair(String tableName, ImmutableBytesWritable rowkey) {
+ super();
+ Preconditions.checkNotNull(tableName);
+ Preconditions.checkNotNull(rowkey);
+ this.tableName = tableName;
+ this.rowkey = rowkey;
+ }
+
+ public String getTableName() {
+ return tableName;
+ }
+
+ public void setTableName(String tableName) {
+ this.tableName = tableName;
+ }
+
+ public ImmutableBytesWritable getRowkey() {
+ return rowkey;
+ }
+
+ public void setRowkey(ImmutableBytesWritable rowkey) {
+ this.rowkey = rowkey;
+ }
+
+ @Override
+ public void readFields(DataInput input) throws IOException {
+ tableName = WritableUtils.readString(input);
+ rowkey = new ImmutableBytesWritable();
+ rowkey.readFields(input);
+ }
+
+ @Override
+ public void write(DataOutput output) throws IOException {
+ WritableUtils.writeString(output,tableName);
+ rowkey.write(output);
+ }
+
+ @Override
+ public int compareTo(TableRowkeyPair other) {
+ String otherTableName = other.getTableName();
+ if(this.tableName.equals(otherTableName)) {
+ return this.rowkey.compareTo(other.getRowkey());
+ } else {
+ return this.tableName.compareTo(otherTableName);
+ }
+ }
+
+ /** Comparator optimized for <code>TableRowkeyPair</code>. */
+ public static class Comparator extends WritableComparator {
+ private BytesWritable.Comparator comparator = new BytesWritable.Comparator();
+
+ public Comparator() {
+ super(TableRowkeyPair.class);
+ }
+
+ @Override
+ public int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2) {
+ try {
+ int vintL1 = WritableUtils.decodeVIntSize(b1[s1]);
+ int vintL2 = WritableUtils.decodeVIntSize(b2[s2]);
+ int strL1 = readVInt(b1, s1);
+ int strL2 = readVInt(b2, s2);
+ int cmp = compareBytes(b1, s1 + vintL1, strL1, b2, s2 + vintL2, strL2);
+ if (cmp != 0) {
+ return cmp;
+ }
+ int vintL3 = WritableUtils.decodeVIntSize(b1[s1 + vintL1 + strL1]);
+ int vintL4 = WritableUtils.decodeVIntSize(b2[s2 + vintL2 + strL2]);
+ int strL3 = readVInt(b1, s1 + vintL1 + strL1);
+ int strL4 = readVInt(b2, s2 + vintL2 + strL2);
+ return comparator.compare(b1, s1 + vintL1 + strL1 + vintL3, strL3, b2, s2
+ + vintL2 + strL2 + vintL4, strL4);
+
+ } catch(Exception ex) {
+ throw new IllegalArgumentException(ex);
+ }
+ }
+ }
+
+ static {
+ WritableComparator.define(TableRowkeyPair.class, new Comparator());
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/578979a1/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/bulkload/TargetTableRef.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/bulkload/TargetTableRef.java b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/bulkload/TargetTableRef.java
new file mode 100644
index 0000000..1a846f9
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/bulkload/TargetTableRef.java
@@ -0,0 +1,70 @@
+/*
+ * 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.bulkload;
+
+import java.util.Map;
+
+import org.codehaus.jackson.annotate.JsonCreator;
+import org.codehaus.jackson.annotate.JsonProperty;
+
+import com.google.common.collect.Maps;
+
+/**
+ * Represents the logical and physical name of a single table to which data is to be loaded.
+ *
+ * This class exists to allow for the difference between HBase physical table names and
+ * Phoenix logical table names.
+ */
+public class TargetTableRef {
+
+ @JsonProperty
+ private final String logicalName;
+
+ @JsonProperty
+ private final String physicalName;
+
+ @JsonProperty
+ private Map<String,String> configuration = Maps.newHashMap();
+
+ public TargetTableRef(String name) {
+ this(name, name);
+ }
+
+ @JsonCreator
+ private TargetTableRef(@JsonProperty("logicalName") String logicalName,
+ @JsonProperty("physicalName") String physicalName) {
+ this.logicalName = logicalName;
+ this.physicalName = physicalName;
+ }
+
+ public String getLogicalName() {
+ return logicalName;
+ }
+
+ public String getPhysicalName() {
+ return physicalName;
+ }
+
+ public Map<String, String> getConfiguration() {
+ return configuration;
+ }
+
+ public void setConfiguration(Map<String, String> configuration) {
+ this.configuration = configuration;
+ }
+}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/578979a1/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/bulkload/TargetTableRefFunctions.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/bulkload/TargetTableRefFunctions.java b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/bulkload/TargetTableRefFunctions.java
new file mode 100644
index 0000000..d786842
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/bulkload/TargetTableRefFunctions.java
@@ -0,0 +1,95 @@
+/*
+ * 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.bulkload;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.codehaus.jackson.map.ObjectMapper;
+
+import com.google.common.base.Function;
+import com.google.common.collect.Lists;
+
+/**
+ * Utility functions to get/put json.
+ *
+ */
+public class TargetTableRefFunctions {
+
+ public static Function<TargetTableRef,String> TO_JSON = new Function<TargetTableRef,String>() {
+
+ @Override
+ public String apply(TargetTableRef input) {
+ try {
+ ObjectMapper mapper = new ObjectMapper();
+ return mapper.writeValueAsString(input);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+
+ }
+ };
+
+ public static Function<String,TargetTableRef> FROM_JSON = new Function<String,TargetTableRef>() {
+
+ @Override
+ public TargetTableRef apply(String json) {
+ try {
+ ObjectMapper mapper = new ObjectMapper();
+ return mapper.readValue(json, TargetTableRef.class);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+
+ }
+ };
+
+ public static Function<List<TargetTableRef>,String> NAMES_TO_JSON = new Function<List<TargetTableRef>,String>() {
+
+ @Override
+ public String apply(List<TargetTableRef> input) {
+ try {
+ List<String> tableNames = Lists.newArrayListWithCapacity(input.size());
+ for(TargetTableRef table : input) {
+ tableNames.add(table.getPhysicalName());
+ }
+ ObjectMapper mapper = new ObjectMapper();
+ return mapper.writeValueAsString(tableNames);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+
+ }
+ };
+
+ public static Function<String,List<String>> NAMES_FROM_JSON = new Function<String,List<String>>() {
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public List<String> apply(String json) {
+ try {
+ ObjectMapper mapper = new ObjectMapper();
+ return mapper.readValue(json, ArrayList.class);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+
+ }
+ };
+ }
http://git-wip-us.apache.org/repos/asf/phoenix/blob/578979a1/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/util/PhoenixConfigurationUtil.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/util/PhoenixConfigurationUtil.java b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/util/PhoenixConfigurationUtil.java
index 8bf786b..9e29fba 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/util/PhoenixConfigurationUtil.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/util/PhoenixConfigurationUtil.java
@@ -23,13 +23,6 @@ import java.sql.SQLException;
import java.util.List;
import java.util.Map;
-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.Iterables;
-import com.google.common.collect.Lists;
-
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
@@ -41,13 +34,17 @@ import org.apache.hadoop.mapreduce.lib.db.DBInputFormat.NullDBWritable;
import org.apache.hadoop.mapreduce.lib.db.DBWritable;
import org.apache.hadoop.util.ReflectionUtils;
import org.apache.phoenix.jdbc.PhoenixConnection;
-import org.apache.phoenix.mapreduce.CsvToKeyValueMapper.DefaultImportPreUpsertKeyValueProcessor;
+import org.apache.phoenix.mapreduce.FormatToKeyValueMapper;
import org.apache.phoenix.mapreduce.ImportPreUpsertKeyValueProcessor;
import org.apache.phoenix.mapreduce.PhoenixInputFormat;
import org.apache.phoenix.util.ColumnInfo;
import org.apache.phoenix.util.PhoenixRuntime;
import org.apache.phoenix.util.QueryUtil;
+import com.google.common.base.Joiner;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
+
import static org.apache.commons.lang.StringUtils.isNotEmpty;
/**
@@ -419,7 +416,7 @@ public final class PhoenixConfigurationUtil {
Class<? extends ImportPreUpsertKeyValueProcessor> processorClass = null;
try {
processorClass = conf.getClass(
- UPSERT_HOOK_CLASS_CONFKEY, DefaultImportPreUpsertKeyValueProcessor.class,
+ UPSERT_HOOK_CLASS_CONFKEY, FormatToKeyValueMapper.DefaultImportPreUpsertKeyValueProcessor.class,
ImportPreUpsertKeyValueProcessor.class);
} catch (Exception e) {
throw new IllegalStateException("Couldn't load upsert hook class", e);
http://git-wip-us.apache.org/repos/asf/phoenix/blob/578979a1/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 b8b284a..cdd9d7b 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
@@ -17,36 +17,25 @@
*/
package org.apache.phoenix.util;
-import com.google.common.base.Charsets;
-import com.google.common.base.Joiner;
-import com.google.common.base.Throwables;
-import com.google.common.collect.ImmutableMap;
-import com.google.common.collect.Lists;
-import com.google.common.collect.Maps;
+import java.io.File;
+import java.io.Reader;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
import org.apache.commons.csv.CSVFormat;
import org.apache.commons.csv.CSVParser;
import org.apache.commons.csv.CSVRecord;
-import org.apache.phoenix.exception.SQLExceptionCode;
-import org.apache.phoenix.exception.SQLExceptionInfo;
import org.apache.phoenix.jdbc.PhoenixConnection;
-import org.apache.phoenix.schema.types.PDataType;
import org.apache.phoenix.util.csv.CsvUpsertExecutor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.io.File;
-import java.io.Reader;
-import java.sql.Connection;
-import java.sql.DatabaseMetaData;
-import java.sql.ResultSet;
-import java.sql.SQLException;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.TreeSet;
+import com.google.common.base.Charsets;
+import com.google.common.base.Joiner;
+import com.google.common.base.Throwables;
+import com.google.common.collect.ImmutableMap;
/***
* Upserts CSV data using Phoenix JDBC connection
@@ -222,7 +211,7 @@ public class CSVCommonsLoader {
long start = System.currentTimeMillis();
CsvUpsertListener upsertListener = new CsvUpsertListener(conn,
conn.getMutateBatchSize(), isStrict);
- CsvUpsertExecutor csvUpsertExecutor = CsvUpsertExecutor.create(conn, tableName,
+ CsvUpsertExecutor csvUpsertExecutor = new CsvUpsertExecutor(conn, tableName,
columnInfoList, upsertListener, arrayElementSeparator);
csvUpsertExecutor.execute(csvParser);
@@ -267,131 +256,10 @@ public class CSVCommonsLoader {
default:
throw new IllegalStateException("parser has unknown column source.");
}
- return generateColumnInfo(conn, tableName, columns, isStrict);
- }
-
- /**
- * 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 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
- */
- public static List<ColumnInfo> generateColumnInfo(Connection conn,
- String tableName, List<String> columns, boolean strict)
- throws SQLException {
- Map<String, Integer> columnNameToTypeMap = Maps.newLinkedHashMap();
- Set<String> ambiguousColumnNames = new HashSet<String>();
- Map<String, Integer> fullColumnNameToTypeMap = 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);
- String[] schemaAndTable = escapedTableName.split("\\.");
- ResultSet rs = null;
- try {
- rs = dbmd.getColumns(null, (schemaAndTable.length == 1 ? ""
- : schemaAndTable[0]),
- (schemaAndTable.length == 1 ? escapedTableName
- : schemaAndTable[1]), null);
- while (rs.next()) {
- String colName = rs.getString(QueryUtil.COLUMN_NAME_POSITION);
- String colFam = rs.getString(QueryUtil.COLUMN_FAMILY_POSITION);
-
- // use family qualifier, if available, otherwise, use column name
- String fullColumn = (colFam==null?colName:String.format("%s.%s",colFam,colName));
- String sqlTypeName = rs.getString(QueryUtil.DATA_TYPE_NAME_POSITION);
-
- // allow for both bare and family qualified names.
- if (columnNameToTypeMap.keySet().contains(colName)) {
- ambiguousColumnNames.add(colName);
- }
- columnNameToTypeMap.put(
- colName,
- PDataType.fromSqlTypeName(sqlTypeName).getSqlType());
- fullColumnNameToTypeMap.put(
- fullColumn,
- PDataType.fromSqlTypeName(sqlTypeName).getSqlType());
- }
- if (columnNameToTypeMap.isEmpty()) {
- throw new IllegalArgumentException("Table " + tableName + " not found");
- }
- } finally {
- if (rs != null) {
- rs.close();
- }
- }
- List<ColumnInfo> columnInfoList = Lists.newArrayList();
- Set<String> unresolvedColumnNames = new TreeSet<String>();
- if (columns == null) {
- // use family qualified names by default, if no columns are specified.
- for (Map.Entry<String, Integer> entry : fullColumnNameToTypeMap
- .entrySet()) {
- columnInfoList.add(new ColumnInfo(entry.getKey(), entry.getValue()));
- }
- } else {
- // Leave "null" as indication to skip b/c it doesn't exist
- for (int i = 0; i < columns.size(); i++) {
- String columnName = columns.get(i).trim();
- Integer sqlType = null;
- if (fullColumnNameToTypeMap.containsKey(columnName)) {
- sqlType = fullColumnNameToTypeMap.get(columnName);
- } else if (columnNameToTypeMap.containsKey(columnName)) {
- if (ambiguousColumnNames.contains(columnName)) {
- unresolvedColumnNames.add(columnName);
- }
- // fall back to bare column name.
- sqlType = columnNameToTypeMap.get(columnName);
- }
- if (unresolvedColumnNames.size()>0) {
- StringBuilder exceptionMessage = new StringBuilder();
- boolean first = true;
- exceptionMessage.append("Unable to resolve these column names to a single column family:\n");
- for (String col : unresolvedColumnNames) {
- if (first) first = false;
- else exceptionMessage.append(",");
- exceptionMessage.append(col);
- }
- exceptionMessage.append("\nAvailable columns with column families:\n");
- first = true;
- for (String col : fullColumnNameToTypeMap.keySet()) {
- if (first) first = false;
- else exceptionMessage.append(",");
- exceptionMessage.append(col);
- }
- throw new SQLException(exceptionMessage.toString());
- }
-
- if (sqlType == null) {
- if (strict) {
- throw new SQLExceptionInfo.Builder(
- SQLExceptionCode.COLUMN_NOT_FOUND)
- .setColumnName(columnName)
- .setTableName(tableName).build()
- .buildException();
- }
- unfoundColumnCount++;
- } else {
- columnInfoList.add(new ColumnInfo(columnName, sqlType));
- }
- }
- if (unfoundColumnCount == columns.size()) {
- throw new SQLExceptionInfo.Builder(
- SQLExceptionCode.COLUMN_NOT_FOUND)
- .setColumnName(
- Arrays.toString(columns.toArray(new String[0])))
- .setTableName(tableName).build().buildException();
- }
- }
- return columnInfoList;
+ return SchemaUtil.generateColumnInfo(conn, tableName, columns, isStrict);
}
- static class CsvUpsertListener implements CsvUpsertExecutor.UpsertListener {
+ static class CsvUpsertListener implements UpsertExecutor.UpsertListener<CSVRecord> {
private final PhoenixConnection conn;
private final int upsertBatchSize;
http://git-wip-us.apache.org/repos/asf/phoenix/blob/578979a1/phoenix-core/src/main/java/org/apache/phoenix/util/SchemaUtil.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/util/SchemaUtil.java b/phoenix-core/src/main/java/org/apache/phoenix/util/SchemaUtil.java
index 1693600..cb8aced 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/util/SchemaUtil.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/util/SchemaUtil.java
@@ -17,21 +17,20 @@
*/
package org.apache.phoenix.util;
-import static com.google.common.base.Preconditions.checkArgument;
-import static com.google.common.base.Preconditions.checkNotNull;
-import static com.google.common.base.Strings.isNullOrEmpty;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_CATALOG_NAME_BYTES;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_FUNCTION_NAME_BYTES;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_STATS_NAME_BYTES;
-
+import java.sql.Connection;
+import java.sql.DatabaseMetaData;
+import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
-
+import java.util.Set;
+import java.util.TreeSet;
import javax.annotation.Nullable;
import org.apache.hadoop.hbase.KeyValue;
@@ -64,6 +63,15 @@ import org.apache.phoenix.schema.types.PVarbinary;
import org.apache.phoenix.schema.types.PVarchar;
import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.base.Preconditions.checkNotNull;
+import static com.google.common.base.Strings.isNullOrEmpty;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_CATALOG_NAME_BYTES;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_FUNCTION_NAME_BYTES;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_STATS_NAME_BYTES;
/**
*
@@ -757,4 +765,125 @@ public class SchemaUtil {
public static byte getSeparatorByte(boolean rowKeyOrderOptimizable, boolean isNullValue, Expression e) {
return getSeparatorByte(rowKeyOrderOptimizable, isNullValue, e.getSortOrder());
}
+
+ /**
+ * 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 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
+ */
+ public static List<ColumnInfo> generateColumnInfo(Connection conn,
+ String tableName, List<String> columns, boolean strict)
+ throws SQLException {
+ Map<String, Integer> columnNameToTypeMap = Maps.newLinkedHashMap();
+ Set<String> ambiguousColumnNames = new HashSet<String>();
+ Map<String, Integer> fullColumnNameToTypeMap = 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);
+ String[] schemaAndTable = escapedTableName.split("\\.");
+ ResultSet rs = null;
+ try {
+ rs = dbmd.getColumns(null, (schemaAndTable.length == 1 ? ""
+ : schemaAndTable[0]),
+ (schemaAndTable.length == 1 ? escapedTableName
+ : schemaAndTable[1]), null);
+ while (rs.next()) {
+ String colName = rs.getString(QueryUtil.COLUMN_NAME_POSITION);
+ String colFam = rs.getString(QueryUtil.COLUMN_FAMILY_POSITION);
+
+ // use family qualifier, if available, otherwise, use column name
+ String fullColumn = (colFam==null?colName:String.format("%s.%s",colFam,colName));
+ String sqlTypeName = rs.getString(QueryUtil.DATA_TYPE_NAME_POSITION);
+
+ // allow for both bare and family qualified names.
+ if (columnNameToTypeMap.keySet().contains(colName)) {
+ ambiguousColumnNames.add(colName);
+ }
+ columnNameToTypeMap.put(
+ colName,
+ PDataType.fromSqlTypeName(sqlTypeName).getSqlType());
+ fullColumnNameToTypeMap.put(
+ fullColumn,
+ PDataType.fromSqlTypeName(sqlTypeName).getSqlType());
+ }
+ if (columnNameToTypeMap.isEmpty()) {
+ throw new IllegalArgumentException("Table " + tableName + " not found");
+ }
+ } finally {
+ if (rs != null) {
+ rs.close();
+ }
+ }
+ List<ColumnInfo> columnInfoList = Lists.newArrayList();
+ Set<String> unresolvedColumnNames = new TreeSet<String>();
+ if (columns == null) {
+ // use family qualified names by default, if no columns are specified.
+ for (Map.Entry<String, Integer> entry : fullColumnNameToTypeMap
+ .entrySet()) {
+ columnInfoList.add(new ColumnInfo(entry.getKey(), entry.getValue()));
+ }
+ } else {
+ // Leave "null" as indication to skip b/c it doesn't exist
+ for (int i = 0; i < columns.size(); i++) {
+ String columnName = columns.get(i).trim();
+ Integer sqlType = null;
+ if (fullColumnNameToTypeMap.containsKey(columnName)) {
+ sqlType = fullColumnNameToTypeMap.get(columnName);
+ } else if (columnNameToTypeMap.containsKey(columnName)) {
+ if (ambiguousColumnNames.contains(columnName)) {
+ unresolvedColumnNames.add(columnName);
+ }
+ // fall back to bare column name.
+ sqlType = columnNameToTypeMap.get(columnName);
+ }
+ if (unresolvedColumnNames.size()>0) {
+ StringBuilder exceptionMessage = new StringBuilder();
+ boolean first = true;
+ exceptionMessage.append("Unable to resolve these column names to a single column family:\n");
+ for (String col : unresolvedColumnNames) {
+ if (first) first = false;
+ else exceptionMessage.append(",");
+ exceptionMessage.append(col);
+ }
+ exceptionMessage.append("\nAvailable columns with column families:\n");
+ first = true;
+ for (String col : fullColumnNameToTypeMap.keySet()) {
+ if (first) first = false;
+ else exceptionMessage.append(",");
+ exceptionMessage.append(col);
+ }
+ throw new SQLException(exceptionMessage.toString());
+ }
+
+ if (sqlType == null) {
+ if (strict) {
+ throw new SQLExceptionInfo.Builder(
+ SQLExceptionCode.COLUMN_NOT_FOUND)
+ .setColumnName(columnName)
+ .setTableName(tableName).build()
+ .buildException();
+ }
+ unfoundColumnCount++;
+ } else {
+ columnInfoList.add(new ColumnInfo(columnName, sqlType));
+ }
+ }
+ if (unfoundColumnCount == columns.size()) {
+ throw new SQLExceptionInfo.Builder(
+ SQLExceptionCode.COLUMN_NOT_FOUND)
+ .setColumnName(
+ Arrays.toString(columns.toArray(new String[0])))
+ .setTableName(tableName).build().buildException();
+ }
+ }
+ return columnInfoList;
+ }
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/578979a1/phoenix-core/src/main/java/org/apache/phoenix/util/UpsertExecutor.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/util/UpsertExecutor.java b/phoenix-core/src/main/java/org/apache/phoenix/util/UpsertExecutor.java
new file mode 100644
index 0000000..d9ce5f2
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/util/UpsertExecutor.java
@@ -0,0 +1,156 @@
+/*
+ * 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 java.io.Closeable;
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.util.List;
+
+import org.apache.phoenix.schema.types.PDataType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Function;
+import com.google.common.collect.Lists;
+
+/**
+ * Executes upsert statements on a provided {@code PreparedStatement} based on incoming
+ * {@code RECORDS}. An {@link UpsertListener} is notified each time the prepared statement
+ * is executed.
+ */
+public abstract class UpsertExecutor<RECORD, FIELD> implements Closeable {
+
+ /**
+ * A listener that is called for events based on incoming JSON data.
+ */
+ public interface UpsertListener<RECORD> {
+
+ /**
+ * Called when an upsert has been sucessfully completed. The given upsertCount is the total number of upserts
+ * completed on the caller up to this point.
+ *
+ * @param upsertCount total number of upserts that have been completed
+ */
+ void upsertDone(long upsertCount);
+
+
+ /**
+ * Called when executing a prepared statement has failed on a given record.
+ *
+ * @param record the JSON record that was being upserted when the error occurred
+ */
+ void errorOnRecord(RECORD record, Throwable throwable);
+ }
+
+ private static final Logger LOG = LoggerFactory.getLogger(UpsertExecutor.class);
+
+ protected final Connection conn;
+ protected final List<ColumnInfo> columnInfos;
+ protected final List<PDataType> dataTypes;
+ protected final List<Function<FIELD, Object>> conversionFunctions;
+ protected final PreparedStatement preparedStatement;
+ protected final UpsertListener<RECORD> upsertListener;
+ protected long upsertCount = 0L;
+ protected boolean initFinished = false; // allow subclasses to finish initialization
+
+ private static PreparedStatement createStatement(Connection conn, String tableName,
+ List<ColumnInfo> columnInfoList) {
+ PreparedStatement preparedStatement;
+ try {
+ String upsertSql = QueryUtil.constructUpsertStatement(tableName, columnInfoList);
+ LOG.info("Upserting SQL data with {}", upsertSql);
+ preparedStatement = conn.prepareStatement(upsertSql);
+ } catch (SQLException e) {
+ throw new RuntimeException(e);
+ }
+ return preparedStatement;
+ }
+
+ /**
+ * Construct with the definition of incoming columns, and the statement upon which upsert
+ * statements are to be performed.
+ */
+ public UpsertExecutor(Connection conn, String tableName,
+ List<ColumnInfo> columnInfoList, UpsertListener<RECORD> upsertListener) {
+ this(conn, columnInfoList, createStatement(conn, tableName, columnInfoList), upsertListener);
+ }
+
+ /** Testing constructor. Do not use in prod. */
+ @VisibleForTesting
+ protected UpsertExecutor(Connection conn, List<ColumnInfo> columnInfoList,
+ PreparedStatement preparedStatement, UpsertListener<RECORD> upsertListener) {
+ this.conn = conn;
+ this.upsertListener = upsertListener;
+ this.columnInfos = columnInfoList;
+ this.preparedStatement = preparedStatement;
+ this.dataTypes = Lists.newArrayList();
+ this.conversionFunctions = Lists.newArrayList();
+ }
+
+ /**
+ * Awkward protocol allows subclass constructors to finish initializing context before
+ * proceeding to record processing.
+ */
+ protected void finishInit() {
+ for (ColumnInfo columnInfo : columnInfos) {
+ PDataType dataType = PDataType.fromTypeId(columnInfo.getSqlType());
+ dataTypes.add(dataType);
+ conversionFunctions.add(createConversionFunction(dataType));
+ }
+ this.initFinished = true;
+ }
+
+ /**
+ * Execute upserts for each JSON record contained in the given iterable, notifying this instance's
+ * {@code UpsertListener} for each completed upsert.
+ *
+ * @param records iterable of JSON records to be upserted
+ */
+ public void execute(Iterable<RECORD> records) {
+ if (!initFinished) {
+ finishInit();
+ }
+ for (RECORD record : records) {
+ execute(record);
+ }
+ }
+
+ /**
+ * Upsert a single record.
+ *
+ * @param record JSON record containing the data to be upserted
+ */
+ protected abstract void execute(RECORD record);
+
+ @Override
+ public void close() throws IOException {
+ try {
+ preparedStatement.close();
+ } catch (SQLException e) {
+ // An exception while closing the prepared statement is most likely a sign of a real problem, so we don't
+ // want to hide it with closeQuietly or something similar
+ throw new RuntimeException(e);
+ }
+ }
+
+ protected abstract Function<FIELD, Object> createConversionFunction(PDataType dataType);
+}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/578979a1/phoenix-core/src/main/java/org/apache/phoenix/util/csv/CsvUpsertExecutor.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/util/csv/CsvUpsertExecutor.java b/phoenix-core/src/main/java/org/apache/phoenix/util/csv/CsvUpsertExecutor.java
index e680f5c..4a3af21 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/util/csv/CsvUpsertExecutor.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/util/csv/CsvUpsertExecutor.java
@@ -17,132 +17,55 @@
*/
package org.apache.phoenix.util.csv;
-import java.io.Closeable;
-import java.io.IOException;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.sql.Types;
import java.util.List;
import java.util.Properties;
-
import javax.annotation.Nullable;
import org.apache.commons.csv.CSVRecord;
-import org.apache.phoenix.jdbc.PhoenixConnection;
import org.apache.phoenix.query.QueryServices;
import org.apache.phoenix.query.QueryServicesOptions;
import org.apache.phoenix.schema.types.PDataType;
import org.apache.phoenix.schema.types.PTimestamp;
import org.apache.phoenix.util.ColumnInfo;
import org.apache.phoenix.util.DateUtil;
-import org.apache.phoenix.util.QueryUtil;
+import org.apache.phoenix.util.UpsertExecutor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Function;
-import com.google.common.collect.Lists;
-/**
- * Executes upsert statements on a provided {@code PreparedStatement} based on incoming CSV records, notifying a
- * listener each time the prepared statement is executed.
- */
-public class CsvUpsertExecutor implements Closeable {
+/** {@link UpsertExecutor} over {@link CSVRecord}s. */
+public class CsvUpsertExecutor extends UpsertExecutor<CSVRecord, String> {
private static final Logger LOG = LoggerFactory.getLogger(CsvUpsertExecutor.class);
- private final String arrayElementSeparator;
- private final Connection conn;
- private final List<PDataType> dataTypes;
- private final List<Function<String,Object>> conversionFunctions;
- private final PreparedStatement preparedStatement;
- private final UpsertListener upsertListener;
- private long upsertCount = 0L;
-
- /**
- * A listener that is called for events based on incoming CSV data.
- */
- public static interface UpsertListener {
-
- /**
- * Called when an upsert has been sucessfully completed. The given upsertCount is the total number of upserts
- * completed on the caller up to this point.
- *
- * @param upsertCount total number of upserts that have been completed
- */
- void upsertDone(long upsertCount);
-
-
- /**
- * Called when executing a prepared statement has failed on a given record.
- *
- * @param csvRecord the CSV record that was being upserted when the error occurred
- */
- void errorOnRecord(CSVRecord csvRecord, Throwable throwable);
- }
-
-
- /**
- * Static constructor method for creating a CsvUpsertExecutor.
- *
- * @param conn Phoenix connection upon which upserts are to be performed
- * @param tableName name of the table in which upserts are to be performed
- * @param columnInfoList description of the columns to be upserted to, in the same order as in the CSV input
- * @param upsertListener listener that will be notified of upserts, can be null
- * @param arrayElementSeparator separator string to delimit string representations of arrays
- * @return the created CsvUpsertExecutor
- */
- public static CsvUpsertExecutor create(PhoenixConnection conn, String tableName, List<ColumnInfo> columnInfoList,
- UpsertListener upsertListener, String arrayElementSeparator) {
- PreparedStatement preparedStatement = null;
- try {
- String upsertSql = QueryUtil.constructUpsertStatement(tableName, columnInfoList);
- LOG.info("Upserting SQL data with {}", upsertSql);
- preparedStatement = conn.prepareStatement(upsertSql);
- } catch (SQLException e) {
- throw new RuntimeException(e);
- }
- return new CsvUpsertExecutor(conn, columnInfoList, preparedStatement, upsertListener,
- arrayElementSeparator);
- }
+ protected final String arrayElementSeparator;
- /**
- * Construct with the definition of incoming columns, and the statement upon which upsert statements
- * are to be performed.
- */
- CsvUpsertExecutor(Connection conn, List<ColumnInfo> columnInfoList, PreparedStatement preparedStatement,
- UpsertListener upsertListener, String arrayElementSeparator) {
- this.conn = conn;
- this.preparedStatement = preparedStatement;
- this.upsertListener = upsertListener;
+ /** Testing constructor. Do not use in prod. */
+ @VisibleForTesting
+ protected CsvUpsertExecutor(Connection conn, List<ColumnInfo> columnInfoList,
+ PreparedStatement stmt, UpsertListener<CSVRecord> upsertListener,
+ String arrayElementSeparator) {
+ super(conn, columnInfoList, stmt, upsertListener);
this.arrayElementSeparator = arrayElementSeparator;
- this.dataTypes = Lists.newArrayList();
- this.conversionFunctions = Lists.newArrayList();
- for (ColumnInfo columnInfo : columnInfoList) {
- PDataType dataType = PDataType.fromTypeId(columnInfo.getSqlType());
- dataTypes.add(dataType);
- conversionFunctions.add(createConversionFunction(dataType));
- }
+ finishInit();
}
- /**
- * Execute upserts for each CSV record contained in the given iterable, notifying this instance's
- * {@code UpsertListener} for each completed upsert.
- *
- * @param csvRecords iterable of CSV records to be upserted
- */
- public void execute(Iterable<CSVRecord> csvRecords) {
- for (CSVRecord csvRecord : csvRecords) {
- execute(csvRecord);
- }
+ public CsvUpsertExecutor(Connection conn, String tableName,
+ List<ColumnInfo> columnInfoList, UpsertListener<CSVRecord> upsertListener,
+ String arrayElementSeparator) {
+ super(conn, tableName, columnInfoList, upsertListener);
+ this.arrayElementSeparator = arrayElementSeparator;
+ finishInit();
}
- /**
- * Upsert a single record.
- *
- * @param csvRecord CSV record containing the data to be upserted
- */
- void execute(CSVRecord csvRecord) {
+ @Override
+ protected void execute(CSVRecord csvRecord) {
try {
if (csvRecord.size() < conversionFunctions.size()) {
String message = String.format("CSV record does not have enough values (has %d, but needs %d)",
@@ -170,17 +93,7 @@ public class CsvUpsertExecutor implements Closeable {
}
@Override
- public void close() throws IOException {
- try {
- preparedStatement.close();
- } catch (SQLException e) {
- // An exception while closing the prepared statement is most likely a sign of a real problem, so we don't
- // want to hide it with closeQuietly or something similar
- throw new RuntimeException(e);
- }
- }
-
- private Function<String, Object> createConversionFunction(PDataType dataType) {
+ protected Function<String, Object> createConversionFunction(PDataType dataType) {
if (dataType.isArrayType()) {
return new ArrayDatatypeConversionFunction(
new StringToArrayConverter(
@@ -201,7 +114,7 @@ public class CsvUpsertExecutor implements Closeable {
private final DateUtil.DateTimeParser dateTimeParser;
SimpleDatatypeConversionFunction(PDataType dataType, Connection conn) {
- Properties props = null;
+ Properties props;
try {
props = conn.getClientInfo();
} catch (SQLException e) {
http://git-wip-us.apache.org/repos/asf/phoenix/blob/578979a1/phoenix-core/src/main/java/org/apache/phoenix/util/json/JsonUpsertExecutor.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/util/json/JsonUpsertExecutor.java b/phoenix-core/src/main/java/org/apache/phoenix/util/json/JsonUpsertExecutor.java
new file mode 100644
index 0000000..bbe0e30
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/util/json/JsonUpsertExecutor.java
@@ -0,0 +1,209 @@
+/*
+ * 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.json;
+
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.sql.Types;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import javax.annotation.Nullable;
+
+import org.apache.phoenix.query.QueryServices;
+import org.apache.phoenix.query.QueryServicesOptions;
+import org.apache.phoenix.schema.types.PDataType;
+import org.apache.phoenix.schema.types.PTimestamp;
+import org.apache.phoenix.util.ColumnInfo;
+import org.apache.phoenix.util.DateUtil;
+import org.apache.phoenix.util.UpsertExecutor;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.CaseFormat;
+import com.google.common.base.Function;
+
+/** {@link UpsertExecutor} over {@link Map} objects, as parsed from JSON. */
+public class JsonUpsertExecutor extends UpsertExecutor<Map<?, ?>, Object> {
+
+ protected static final Logger LOG = LoggerFactory.getLogger(JsonUpsertExecutor.class);
+
+ /** Testing constructor. Do not use in prod. */
+ @VisibleForTesting
+ protected JsonUpsertExecutor(Connection conn, List<ColumnInfo> columnInfoList,
+ PreparedStatement stmt, UpsertListener<Map<?, ?>> upsertListener) {
+ super(conn, columnInfoList, stmt, upsertListener);
+ finishInit();
+ }
+
+ public JsonUpsertExecutor(Connection conn, String tableName, List<ColumnInfo> columnInfoList,
+ UpsertExecutor.UpsertListener<Map<?, ?>> upsertListener) {
+ super(conn, tableName, columnInfoList, upsertListener);
+ finishInit();
+ }
+
+ @Override
+ protected void execute(Map<?, ?> record) {
+ int fieldIndex = 0;
+ String colName = null;
+ try {
+ if (record.size() < conversionFunctions.size()) {
+ String message = String.format("JSON record does not have enough values (has %d, but needs %d)",
+ record.size(), conversionFunctions.size());
+ throw new IllegalArgumentException(message);
+ }
+ for (fieldIndex = 0; fieldIndex < conversionFunctions.size(); fieldIndex++) {
+ colName = CaseFormat.UPPER_UNDERSCORE.to(
+ CaseFormat.LOWER_UNDERSCORE, columnInfos.get(fieldIndex).getColumnName());
+ if (colName.contains(".")) {
+ StringBuilder sb = new StringBuilder();
+ String[] parts = colName.split("\\.");
+ // assume first part is the column family name; omita
+ for (int i = 1; i < parts.length; i++) {
+ sb.append(parts[i]);
+ if (i != parts.length - 1) {
+ sb.append(".");
+ }
+ }
+ colName = sb.toString();
+ }
+ if (colName.contains("\"")) {
+ colName = colName.replace("\"", "");
+ }
+ Object sqlValue = conversionFunctions.get(fieldIndex).apply(record.get(colName));
+ if (sqlValue != null) {
+ preparedStatement.setObject(fieldIndex + 1, sqlValue);
+ } else {
+ preparedStatement.setNull(fieldIndex + 1, dataTypes.get(fieldIndex).getSqlType());
+ }
+ }
+ preparedStatement.execute();
+ upsertListener.upsertDone(++upsertCount);
+ } catch (Exception e) {
+ if (LOG.isDebugEnabled()) {
+ // Even though this is an error we only log it with debug logging because we're notifying the
+ // listener, and it can do its own logging if needed
+ LOG.debug("Error on record " + record + ", fieldIndex " + fieldIndex + ", colName " + colName, e);
+ }
+ upsertListener.errorOnRecord(record, new Exception("fieldIndex: " + fieldIndex + ", colName " + colName, e));
+ }
+ }
+
+ @Override
+ public void close() throws IOException {
+ try {
+ preparedStatement.close();
+ } catch (SQLException e) {
+ // An exception while closing the prepared statement is most likely a sign of a real problem, so we don't
+ // want to hide it with closeQuietly or something similar
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ protected Function<Object, Object> createConversionFunction(PDataType dataType) {
+ if (dataType.isArrayType()) {
+ return new ArrayDatatypeConversionFunction(
+ new ObjectToArrayConverter(
+ conn,
+ PDataType.fromTypeId(dataType.getSqlType() - PDataType.ARRAY_TYPE_BASE)));
+ } else {
+ return new SimpleDatatypeConversionFunction(dataType, this.conn);
+ }
+ }
+
+ /**
+ * Performs typed conversion from String values to a given column value type.
+ */
+ static class SimpleDatatypeConversionFunction implements Function<Object, Object> {
+
+ private final PDataType dataType;
+ private final DateUtil.DateTimeParser dateTimeParser;
+
+ SimpleDatatypeConversionFunction(PDataType dataType, Connection conn) {
+ Properties props;
+ try {
+ props = conn.getClientInfo();
+ } catch (SQLException e) {
+ throw new RuntimeException(e);
+ }
+ this.dataType = dataType;
+ if (dataType.isCoercibleTo(PTimestamp.INSTANCE)) {
+ // TODO: move to DateUtil
+ String dateFormat;
+ int dateSqlType = dataType.getResultSetSqlType();
+ if (dateSqlType == Types.DATE) {
+ dateFormat = props.getProperty(QueryServices.DATE_FORMAT_ATTRIB,
+ DateUtil.DEFAULT_DATE_FORMAT);
+ } else if (dateSqlType == Types.TIME) {
+ dateFormat = props.getProperty(QueryServices.TIME_FORMAT_ATTRIB,
+ DateUtil.DEFAULT_TIME_FORMAT);
+ } else {
+ dateFormat = props.getProperty(QueryServices.TIMESTAMP_FORMAT_ATTRIB,
+ DateUtil.DEFAULT_TIMESTAMP_FORMAT);
+ }
+ String timeZoneId = props.getProperty(QueryServices.DATE_FORMAT_TIMEZONE_ATTRIB,
+ QueryServicesOptions.DEFAULT_DATE_FORMAT_TIMEZONE);
+ this.dateTimeParser = DateUtil.getDateTimeParser(dateFormat, dataType, timeZoneId);
+ } else {
+ this.dateTimeParser = null;
+ }
+ }
+
+ @Nullable
+ @Override
+ public Object apply(@Nullable Object input) {
+ if (input == null) {
+ return null;
+ }
+ if (dateTimeParser != null && input instanceof String) {
+ final String s = (String) input;
+ long epochTime = dateTimeParser.parseDateTime(s);
+ byte[] byteValue = new byte[dataType.getByteSize()];
+ dataType.getCodec().encodeLong(epochTime, byteValue, 0);
+ return dataType.toObject(byteValue);
+ }
+ return dataType.toObject(input, dataType);
+ }
+ }
+
+ /**
+ * Converts string representations of arrays into Phoenix arrays of the correct type.
+ */
+ private static class ArrayDatatypeConversionFunction implements Function<Object, Object> {
+
+ private final ObjectToArrayConverter arrayConverter;
+
+ private ArrayDatatypeConversionFunction(ObjectToArrayConverter arrayConverter) {
+ this.arrayConverter = arrayConverter;
+ }
+
+ @Nullable
+ @Override
+ public Object apply(@Nullable Object input) {
+ try {
+ return arrayConverter.toArray(input);
+ } catch (SQLException e) {
+ throw new RuntimeException(e);
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/578979a1/phoenix-core/src/main/java/org/apache/phoenix/util/json/ObjectToArrayConverter.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/util/json/ObjectToArrayConverter.java b/phoenix-core/src/main/java/org/apache/phoenix/util/json/ObjectToArrayConverter.java
new file mode 100644
index 0000000..16bef15
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/util/json/ObjectToArrayConverter.java
@@ -0,0 +1,69 @@
+/*
+ * 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.json;
+
+import java.sql.Array;
+import java.sql.Connection;
+import java.sql.SQLException;
+import java.util.List;
+
+import org.apache.phoenix.schema.types.PDataType;
+
+import com.google.common.collect.Iterables;
+import com.google.common.collect.Lists;
+
+/**
+ * Converts Objects (presumably lists) into Phoenix arrays.
+ */
+class ObjectToArrayConverter {
+
+ private final Connection conn;
+ private final PDataType elementDataType;
+ private final JsonUpsertExecutor.SimpleDatatypeConversionFunction elementConvertFunction;
+
+ /**
+ * Instantiate with the array value separator and data type.
+ *
+ * @param conn Phoenix connection to target database
+ * @param elementDataType datatype of the elements of arrays to be created
+ */
+ public ObjectToArrayConverter(Connection conn, PDataType elementDataType) {
+ this.conn = conn;
+ this.elementDataType = elementDataType;
+ this.elementConvertFunction =
+ new JsonUpsertExecutor.SimpleDatatypeConversionFunction(elementDataType, this.conn);
+ }
+
+ /**
+ * Convert an input delimited string into a phoenix array of the configured type.
+ *
+ * @param input string containing delimited array values
+ * @return the array containing the values represented in the input string
+ */
+ public Array toArray(Object input) throws SQLException {
+ if (input == null) {
+ return conn.createArrayOf(elementDataType.getSqlTypeName(), new Object[0]);
+ }
+ List<?> list = (List<?>) input;
+ if (list.isEmpty()) {
+ return conn.createArrayOf(elementDataType.getSqlTypeName(), new Object[0]);
+ }
+ return conn.createArrayOf(elementDataType.getSqlTypeName(),
+ Lists.newArrayList(Iterables.transform(list, elementConvertFunction)).toArray());
+ }
+}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/578979a1/phoenix-core/src/test/java/org/apache/phoenix/mapreduce/BulkLoadToolTest.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/test/java/org/apache/phoenix/mapreduce/BulkLoadToolTest.java b/phoenix-core/src/test/java/org/apache/phoenix/mapreduce/BulkLoadToolTest.java
new file mode 100644
index 0000000..95e9b43
--- /dev/null
+++ b/phoenix-core/src/test/java/org/apache/phoenix/mapreduce/BulkLoadToolTest.java
@@ -0,0 +1,78 @@
+/*
+ * 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 java.util.Arrays;
+import java.util.Collection;
+
+import org.apache.commons.cli.CommandLine;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import static org.junit.Assert.assertEquals;
+
+@RunWith(Parameterized.class)
+public class BulkLoadToolTest {
+
+ @Parameterized.Parameters
+ public static Collection<Object[]> params() {
+ return Arrays.asList(new Object[][]{
+ { new CsvBulkLoadTool() },
+ { new JsonBulkLoadTool() },
+ });
+ }
+
+ @Parameterized.Parameter(value = 0)
+ public AbstractBulkLoadTool bulkLoadTool;
+
+ @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"));
+ }
+}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/578979a1/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
index f52a837..3c6271a 100644
--- a/phoenix-core/src/test/java/org/apache/phoenix/mapreduce/CsvBulkImportUtilTest.java
+++ b/phoenix-core/src/test/java/org/apache/phoenix/mapreduce/CsvBulkImportUtilTest.java
@@ -17,9 +17,6 @@
*/
package org.apache.phoenix.mapreduce;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNull;
-
import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
@@ -29,11 +26,10 @@ import java.util.List;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.phoenix.mapreduce.util.PhoenixConfigurationUtil;
-import org.apache.phoenix.schema.types.PInteger;
-import org.apache.phoenix.util.ColumnInfo;
import org.junit.Test;
-import com.google.common.collect.ImmutableList;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
public class CsvBulkImportUtilTest {
@@ -41,16 +37,11 @@ public class CsvBulkImportUtilTest {
public void testInitCsvImportJob() throws IOException {
Configuration conf = new Configuration();
- String tableName = "SCHEMANAME.TABLENAME";
char delimiter = '\001';
char quote = '\002';
char escape = '!';
- List<ColumnInfo> columnInfoList = ImmutableList.of(
- new ColumnInfo("MYCOL", PInteger.INSTANCE.getSqlType()));
-
- CsvBulkImportUtil.initCsvImportJob(
- conf, tableName, delimiter, quote, escape, null, columnInfoList, true);
+ CsvBulkImportUtil.initCsvImportJob(conf, delimiter, quote, escape, null);
// Serialize and deserialize the config to ensure that there aren't any issues
// with non-printable characters as delimiters
@@ -61,7 +52,6 @@ public class CsvBulkImportUtilTest {
Configuration deserialized = new Configuration();
deserialized.addResource(new FileInputStream(tempFile));
- assertEquals(tableName, deserialized.get(CsvToKeyValueMapper.TABLE_NAME_CONFKEY));
assertEquals(Character.valueOf('\001'),
CsvBulkImportUtil.getCharacter(deserialized, CsvToKeyValueMapper.FIELD_DELIMITER_CONFKEY));
assertEquals(Character.valueOf('\002'),
@@ -69,8 +59,6 @@ public class CsvBulkImportUtilTest {
assertEquals(Character.valueOf('!'),
CsvBulkImportUtil.getCharacter(deserialized, CsvToKeyValueMapper.ESCAPE_CHAR_CONFKEY));
assertNull(deserialized.get(CsvToKeyValueMapper.ARRAY_DELIMITER_CONFKEY));
- assertEquals(columnInfoList, CsvToKeyValueMapper.buildColumnInfoList(deserialized));
- assertEquals(true, deserialized.getBoolean(CsvToKeyValueMapper.IGNORE_INVALID_ROW_CONFKEY, false));
tempFile.delete();
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/578979a1/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
deleted file mode 100644
index 33bb976..0000000
--- a/phoenix-core/src/test/java/org/apache/phoenix/mapreduce/CsvBulkLoadToolTest.java
+++ /dev/null
@@ -1,69 +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.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"));
- }
-}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/578979a1/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
index dc6f497..fe4e068 100644
--- a/phoenix-core/src/test/java/org/apache/phoenix/mapreduce/CsvToKeyValueMapperTest.java
+++ b/phoenix-core/src/test/java/org/apache/phoenix/mapreduce/CsvToKeyValueMapperTest.java
@@ -17,25 +17,13 @@
*/
package org.apache.phoenix.mapreduce;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertTrue;
-
import java.io.IOException;
-import java.util.List;
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.mapreduce.util.PhoenixConfigurationUtil;
-import org.apache.phoenix.schema.types.PInteger;
-import org.apache.phoenix.schema.types.PIntegerArray;
-import org.apache.phoenix.schema.types.PUnsignedInt;
-import org.apache.phoenix.util.ColumnInfo;
import org.junit.Test;
-import com.google.common.collect.ImmutableList;
-import com.google.common.collect.Lists;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
public class CsvToKeyValueMapperTest {
@@ -62,72 +50,4 @@ public class CsvToKeyValueMapperTest {
assertTrue(parsed.isConsistent());
assertEquals(1, parsed.getRecordNumber());
}
-
-
- @Test
- public void testBuildColumnInfoList() {
- List<ColumnInfo> columnInfoList = ImmutableList.of(
- new ColumnInfo("idCol", PInteger.INSTANCE.getSqlType()),
- new ColumnInfo("unsignedIntCol", PUnsignedInt.INSTANCE.getSqlType()),
- new ColumnInfo("stringArrayCol", PIntegerArray.INSTANCE.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", PInteger.INSTANCE.getSqlType()),
- null,
- new ColumnInfo("unsignedIntCol", PUnsignedInt.INSTANCE.getSqlType()),
- new ColumnInfo("stringArrayCol", PIntegerArray.INSTANCE.getSqlType()));
-
- Configuration conf = new Configuration();
- CsvToKeyValueMapper.configureColumnInfoList(conf, columnInfoListWithNull);
- List<ColumnInfo> fromConfig = CsvToKeyValueMapper.buildColumnInfoList(conf);
-
- assertEquals(columnInfoListWithNull, fromConfig);
- }
-
- @Test
- public void testLoadPreUpdateProcessor() {
- Configuration conf = new Configuration();
- conf.setClass(PhoenixConfigurationUtil.UPSERT_HOOK_CLASS_CONFKEY, MockUpsertProcessor.class,
- ImportPreUpsertKeyValueProcessor.class);
-
- ImportPreUpsertKeyValueProcessor processor = PhoenixConfigurationUtil.loadPreUpsertProcessor(conf);
- assertEquals(MockUpsertProcessor.class, processor.getClass());
- }
-
- @Test
- public void testLoadPreUpdateProcessor_NotConfigured() {
-
- Configuration conf = new Configuration();
- ImportPreUpsertKeyValueProcessor processor = PhoenixConfigurationUtil.loadPreUpsertProcessor(conf);
-
- assertEquals(CsvToKeyValueMapper.DefaultImportPreUpsertKeyValueProcessor.class,
- processor.getClass());
- }
-
- @Test(expected=IllegalStateException.class)
- public void testLoadPreUpdateProcessor_ClassNotFound() {
- Configuration conf = new Configuration();
- conf.set(PhoenixConfigurationUtil.UPSERT_HOOK_CLASS_CONFKEY, "MyUndefinedClass");
-
- PhoenixConfigurationUtil.loadPreUpsertProcessor(conf);
- }
-
-
- static class MockUpsertProcessor implements ImportPreUpsertKeyValueProcessor {
-
- @Override
- public List<KeyValue> preUpsert(byte[] rowKey, List<KeyValue> keyValues) {
- throw new UnsupportedOperationException("Not yet implemented");
- }
- }
}