You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@phoenix.apache.org by sa...@apache.org on 2016/01/07 00:17:22 UTC
[09/50] [abbrv] phoenix git commit: PHOENIX-2517 Bulk load tools
should support multiple input files
PHOENIX-2517 Bulk load tools should support multiple input files
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/f4257c6a
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/f4257c6a
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/f4257c6a
Branch: refs/heads/4.x-HBase-1.0
Commit: f4257c6a5c7ec8d32927b2d9987d5402740bfc47
Parents: 42506f4
Author: Nick Dimiduk <nd...@apache.org>
Authored: Fri Dec 11 12:41:49 2015 -0800
Committer: Nick Dimiduk <nd...@apache.org>
Committed: Tue Dec 15 12:36:36 2015 -0800
----------------------------------------------------------------------
.../phoenix/mapreduce/CsvBulkLoadToolIT.java | 41 ++++++++++++++++++++
.../phoenix/mapreduce/AbstractBulkLoadTool.java | 16 ++++----
.../mapreduce/MultiHfileOutputFormat.java | 2 +-
3 files changed, 50 insertions(+), 9 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/f4257c6a/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 4a440d6..7daacb4 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
@@ -36,6 +36,7 @@ import org.apache.phoenix.util.DateUtil;
import org.apache.phoenix.util.PhoenixRuntime;
import org.junit.AfterClass;
import org.junit.BeforeClass;
+import org.junit.Ignore;
import org.junit.Test;
import org.junit.experimental.categories.Category;
@@ -166,6 +167,46 @@ public class CsvBulkLoadToolIT {
}
@Test
+ public void testMultipleInputFiles() throws Exception {
+
+ Statement stmt = conn.createStatement();
+ stmt.execute("CREATE TABLE TABLE7 (ID INTEGER NOT NULL PRIMARY KEY, NAME VARCHAR, T DATE) SPLIT ON (1,2)");
+
+ FileSystem fs = FileSystem.get(hbaseTestUtil.getConfiguration());
+ FSDataOutputStream outputStream = fs.create(new Path("/tmp/input1.csv"));
+ PrintWriter printWriter = new PrintWriter(outputStream);
+ printWriter.println("1,Name 1,1970/01/01");
+ printWriter.close();
+ outputStream = fs.create(new Path("/tmp/input2.csv"));
+ printWriter = new PrintWriter(outputStream);
+ printWriter.println("2,Name 2,1970/01/02");
+ printWriter.close();
+
+ CsvBulkLoadTool csvBulkLoadTool = new CsvBulkLoadTool();
+ csvBulkLoadTool.setConf(new Configuration(hbaseTestUtil.getConfiguration()));
+ csvBulkLoadTool.getConf().set(DATE_FORMAT_ATTRIB,"yyyy/MM/dd");
+ int exitCode = csvBulkLoadTool.run(new String[] {
+ "--input", "/tmp/input1.csv,/tmp/input2.csv",
+ "--table", "table7",
+ "--zookeeper", zkQuorum});
+ assertEquals(0, exitCode);
+
+ ResultSet rs = stmt.executeQuery("SELECT id, name, t FROM table7 ORDER BY id");
+ assertTrue(rs.next());
+ assertEquals(1, rs.getInt(1));
+ assertEquals("Name 1", rs.getString(2));
+ assertEquals(DateUtil.parseDate("1970-01-01"), rs.getDate(3));
+ assertTrue(rs.next());
+ assertEquals(2, rs.getInt(1));
+ assertEquals("Name 2", rs.getString(2));
+ assertEquals(DateUtil.parseDate("1970-01-02"), rs.getDate(3));
+ assertFalse(rs.next());
+
+ rs.close();
+ stmt.close();
+ }
+
+ @Test
public void testImportWithIndex() throws Exception {
Statement stmt = conn.createStatement();
http://git-wip-us.apache.org/repos/asf/phoenix/blob/f4257c6a/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
index cf9ddef..1d2594d 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/AbstractBulkLoadTool.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/AbstractBulkLoadTool.java
@@ -74,7 +74,7 @@ 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 INPUT_PATH_OPT = new Option("i", "input", true, "Input path(s) (comma-separated, 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)");
@@ -219,7 +219,7 @@ public abstract class AbstractBulkLoadTool extends Configured implements Tool {
conn.close();
}
- final Path inputPath = new Path(cmdLine.getOptionValue(INPUT_PATH_OPT.getOpt()));
+ final String inputPaths = 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()));
@@ -249,18 +249,18 @@ public abstract class AbstractBulkLoadTool extends Configured implements Tool {
tablesToBeLoaded.add(targetIndexRef);
}
- return submitJob(conf, tableName, inputPath, outputPath, tablesToBeLoaded);
+ return submitJob(conf, tableName, inputPaths, 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) {
+ public int submitJob(final Configuration conf, final String qualifiedTableName,
+ final String inputPaths, final Path outputPath, List<TargetTableRef> tablesToBeLoaded) {
try {
Job job = new Job(conf, "Phoenix MapReduce import for " + qualifiedTableName);
- FileInputFormat.addInputPath(job, inputPath);
+ FileInputFormat.addInputPaths(job, inputPaths);
FileOutputFormat.setOutputPath(job, outputPath);
job.setInputFormatClass(TextInputFormat.class);
@@ -278,7 +278,7 @@ public abstract class AbstractBulkLoadTool extends Configured implements Tool {
// give subclasses their hook
setupJob(job);
- LOG.info("Running MapReduce import job from {} to {}", inputPath, outputPath);
+ LOG.info("Running MapReduce import job from {} to {}", inputPaths, outputPath);
boolean success = job.waitForCompletion(true);
if (success) {
@@ -292,7 +292,7 @@ public abstract class AbstractBulkLoadTool extends Configured implements Tool {
}
return 0;
} catch(Exception e) {
- LOG.error("Error {} occurred submitting BulkLoad ",e.getMessage());
+ LOG.error("Error occurred submitting BulkLoad ", e);
return -1;
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/f4257c6a/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 45a014d..525e79a 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
@@ -687,7 +687,7 @@ public class MultiHfileOutputFormat extends FileOutputFormat<TableRowkeyPair, Ce
conf.set(tableName, tableDefns);
TargetTableRef tbl = TargetTableRefFunctions.FROM_JSON.apply(tableDefns);
- LOG.error(" the table logical name is "+ tbl.getLogicalName());
+ LOG.info(" the table logical name is "+ tbl.getLogicalName());
}
}