You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pig.apache.org by ro...@apache.org on 2013/05/06 04:54:43 UTC
svn commit: r1479460 - in /pig/trunk: CHANGES.txt
contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/avro/AvroStorageUtils.java
contrib/piggybank/java/src/test/java/org/apache/pig/piggybank/test/storage/avro/TestAvroStorage.java
Author: rohini
Date: Mon May 6 02:54:42 2013
New Revision: 1479460
URL: http://svn.apache.org/r1479460
Log:
PIG-3223: AvroStorage does not handle comma separated input paths (dreambird via rohini)
Modified:
pig/trunk/CHANGES.txt
pig/trunk/contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/avro/AvroStorageUtils.java
pig/trunk/contrib/piggybank/java/src/test/java/org/apache/pig/piggybank/test/storage/avro/TestAvroStorage.java
Modified: pig/trunk/CHANGES.txt
URL: http://svn.apache.org/viewvc/pig/trunk/CHANGES.txt?rev=1479460&r1=1479459&r2=1479460&view=diff
==============================================================================
--- pig/trunk/CHANGES.txt (original)
+++ pig/trunk/CHANGES.txt Mon May 6 02:54:42 2013
@@ -316,6 +316,22 @@ PIG-2885: TestJobSumission and TestHBase
PIG-2928: Fix e2e test failures in trunk: FilterBoolean_23/24 (cheolsoo via dvryaboy)
+
+Release 0.11.2 (Unreleased)
+
+INCOMPATIBLE CHANGES
+
+IMPROVEMENTS
+
+OPTIMIZATIONS
+
+PIG-2769: a simple logic causes very long compiling time on pig 0.10.0 (njw45 via dvryaboy) (prev. applied to 0.12)
+
+BUG FIXES
+
+PIG-3223: AvroStorage does not handle comma separated input paths (dreambird via rohini)
+
+
Release 0.11.1 (unreleased)
INCOMPATIBLE CHANGES
Modified: pig/trunk/contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/avro/AvroStorageUtils.java
URL: http://svn.apache.org/viewvc/pig/trunk/contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/avro/AvroStorageUtils.java?rev=1479460&r1=1479459&r2=1479460&view=diff
==============================================================================
--- pig/trunk/contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/avro/AvroStorageUtils.java (original)
+++ pig/trunk/contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/avro/AvroStorageUtils.java Mon May 6 02:54:42 2013
@@ -39,6 +39,7 @@ import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.PathFilter;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
+import org.apache.pig.LoadFunc;
import org.apache.pig.ResourceSchema;
import org.apache.pig.ResourceSchema.ResourceFieldSchema;
import org.apache.pig.data.DataType;
@@ -112,28 +113,38 @@ public class AvroStorageUtils {
/**
* Adds all non-hidden directories and subdirectories to set param
+ * it supports comma-separated input paths and glob style path
*
* @throws IOException
*/
- public static boolean getAllSubDirs(Path path, Configuration conf, Set<Path> paths) throws IOException {
- FileSystem fs = FileSystem.get(path.toUri(), conf);
- FileStatus[] matchedFiles = fs.globStatus(path, PATH_FILTER);
- if (matchedFiles == null || matchedFiles.length == 0) {
- return false;
- }
- for (FileStatus file : matchedFiles) {
- if (file.isDir()) {
- for (FileStatus sub : fs.listStatus(file.getPath())) {
- getAllSubDirs(sub.getPath(), conf, paths);
- }
- } else {
- AvroStorageLog.details("Add input file:" + file);
- paths.add(file.getPath());
+ public static boolean getAllSubDirs(Path path, Configuration conf,
+ Set<Path> paths) throws IOException {
+ String[] pathStrs = LoadFunc.getPathStrings(path.toString());
+ for (String pathStr : pathStrs) {
+ FileSystem fs = FileSystem.get(new Path(pathStr).toUri(), conf);
+ FileStatus[] matchedFiles = fs.globStatus(new Path(pathStr), PATH_FILTER);
+ if (matchedFiles == null || matchedFiles.length == 0) {
+ return false;
+ }
+ for (FileStatus file : matchedFiles) {
+ getAllSubDirsInternal(file, conf, paths, fs);
}
}
return true;
}
+ private static void getAllSubDirsInternal(FileStatus file, Configuration conf,
+ Set<Path> paths, FileSystem fs) throws IOException {
+ if (file.isDir()) {
+ for (FileStatus sub : fs.listStatus(file.getPath())) {
+ getAllSubDirsInternal(sub, conf, paths, fs);
+ }
+ } else {
+ AvroStorageLog.details("Add input file:" + file);
+ paths.add(file.getPath());
+ }
+ }
+
/** check whether there is NO directory in the input file (status) list*/
public static boolean noDir(FileStatus [] ss) {
for (FileStatus s : ss) {
Modified: pig/trunk/contrib/piggybank/java/src/test/java/org/apache/pig/piggybank/test/storage/avro/TestAvroStorage.java
URL: http://svn.apache.org/viewvc/pig/trunk/contrib/piggybank/java/src/test/java/org/apache/pig/piggybank/test/storage/avro/TestAvroStorage.java?rev=1479460&r1=1479459&r2=1479460&view=diff
==============================================================================
--- pig/trunk/contrib/piggybank/java/src/test/java/org/apache/pig/piggybank/test/storage/avro/TestAvroStorage.java (original)
+++ pig/trunk/contrib/piggybank/java/src/test/java/org/apache/pig/piggybank/test/storage/avro/TestAvroStorage.java Mon May 6 02:54:42 2013
@@ -27,6 +27,7 @@ import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.PathFilter;
import org.apache.hadoop.mapreduce.Job;
import org.apache.pig.ExecType;
+import org.apache.pig.LoadFunc;
import org.apache.pig.PigServer;
import org.apache.pig.backend.executionengine.ExecException;
import org.apache.pig.backend.executionengine.ExecJob;
@@ -36,6 +37,7 @@ import org.apache.pig.impl.io.FileLocali
import org.apache.pig.impl.logicalLayer.FrontendException;
import org.apache.pig.piggybank.storage.avro.AvroStorage;
import org.apache.pig.piggybank.storage.avro.PigSchema2Avro;
+import org.apache.pig.test.MiniCluster;
import org.apache.pig.test.Util;
import org.junit.AfterClass;
import org.junit.Assert;
@@ -44,6 +46,7 @@ import org.junit.Test;
import java.io.File;
import java.io.IOException;
+import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Properties;
@@ -70,7 +73,19 @@ public class TestAvroStorage {
};
private static String getInputFile(String file) {
- return "file://" + System.getProperty("user.dir") + "/" + basedir + file;
+ String locations[] = LoadFunc.getPathStrings(file);
+ if (locations.length == 1)
+ return "file://" + System.getProperty("user.dir") + "/" + basedir
+ + file;
+ else {
+ ArrayList<String> pathStrings = new ArrayList<String>();
+ for (int index = 0; index < locations.length; index++) {
+ String f = "file://" + System.getProperty("user.dir") + "/"
+ + basedir + locations[index].trim();
+ pathStrings.add(f);
+ }
+ return LoadFunc.join(pathStrings, ",");
+ }
}
final private String testDir1 = getInputFile("test_dir1");
@@ -79,6 +94,8 @@ public class TestAvroStorage {
final private String testDir1Files321 = getInputFile("test_dir1/test_glob{3,2,1}.avro");
final private String testDir12AllFiles = getInputFile("{test_dir1,test_dir2}/test_glob*.avro");
final private String testDir21AllFiles = getInputFile("{test_dir2,test_dir1}/test_glob*.avro");
+ final private String testCommaSeparated1 = getInputFile("test_dir1/test_glob1.avro,test_dir1/test_glob2.avro,test_dir1/test_glob3.avro");
+ final private String testCommaSeparated2 = getInputFile("test_dir1/test_glob*,test_dir2/test_glob4.avro,test_dir2/test_glob5.avro");
final private String testNoMatchedFiles = getInputFile("test_dir{1,2}/file_that_does_not_exist*.avro");
final private String testArrayFile = getInputFile("test_array.avro");
final private String testArraySchema = getInputFile("test_array.avsc");
@@ -678,6 +695,36 @@ public class TestAvroStorage {
}
@Test
+ public void testComma1() throws IOException {
+ // Verify that comma-separated file can be processed
+ String output = outbasedir + "testComma1";
+ String expected = basedir + "expected_test_dir_1.avro";
+ deleteDirectory(new File(output));
+ String [] queries = {
+ " in = LOAD '" + testCommaSeparated1 + "' USING org.apache.pig.piggybank.storage.avro.AvroStorage ();",
+ " STORE in INTO '" + output + "' USING org.apache.pig.piggybank.storage.avro.AvroStorage (" +
+ " 'schema', '{\"type\":\"array\",\"items\":\"float\"}' );"
+ };
+ testAvroStorage(queries);
+ verifyResults(output, expected);
+ }
+
+ @Test
+ public void testComma2() throws IOException {
+ // Verify that comma-separated file can be processed
+ String output = outbasedir + "testComma2";
+ String expected = basedir + "expected_test_dir_1_2.avro";
+ deleteDirectory(new File(output));
+ String [] queries = {
+ " in = LOAD '" + testCommaSeparated2 + "' USING org.apache.pig.piggybank.storage.avro.AvroStorage ();",
+ " STORE in INTO '" + output + "' USING org.apache.pig.piggybank.storage.avro.AvroStorage (" +
+ " 'schema', '{\"type\":\"array\",\"items\":\"float\"}' );"
+ };
+ testAvroStorage(queries);
+ verifyResults(output, expected);
+ }
+
+ @Test
public void testArrayDefault() throws IOException {
String output= outbasedir + "testArrayDefault";
String expected = basedir + "expected_testArrayDefault.avro";