You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pig.apache.org by ya...@apache.org on 2010/02/03 22:08:18 UTC
svn commit: r906244 - in /hadoop/pig/branches/branch-0.6/contrib/zebra:
CHANGES.txt src/java/org/apache/hadoop/zebra/pig/TableStorer.java
src/test/org/apache/hadoop/zebra/pig/TestTableSortStorerDesc.java
Author: yanz
Date: Wed Feb 3 21:08:18 2010
New Revision: 906244
URL: http://svn.apache.org/viewvc?rev=906244&view=rev
Log:
PIG-1206 Storing descendingly sorted PIG table as unsorted table (yanz)
Added:
hadoop/pig/branches/branch-0.6/contrib/zebra/src/test/org/apache/hadoop/zebra/pig/TestTableSortStorerDesc.java
Modified:
hadoop/pig/branches/branch-0.6/contrib/zebra/CHANGES.txt
hadoop/pig/branches/branch-0.6/contrib/zebra/src/java/org/apache/hadoop/zebra/pig/TableStorer.java
Modified: hadoop/pig/branches/branch-0.6/contrib/zebra/CHANGES.txt
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/branch-0.6/contrib/zebra/CHANGES.txt?rev=906244&r1=906243&r2=906244&view=diff
==============================================================================
--- hadoop/pig/branches/branch-0.6/contrib/zebra/CHANGES.txt (original)
+++ hadoop/pig/branches/branch-0.6/contrib/zebra/CHANGES.txt Wed Feb 3 21:08:18 2010
@@ -6,6 +6,8 @@
IMPROVEMENTS
+ PIG-1206 Storing descendingly sorted PIG table as unsorted table (yanz)
+
PIG-1125 Map/Reduce API Changes (Chao Wang via yanz)
PIG-1104 Streaming Support (Chao Wang via yanz)
Modified: hadoop/pig/branches/branch-0.6/contrib/zebra/src/java/org/apache/hadoop/zebra/pig/TableStorer.java
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/branch-0.6/contrib/zebra/src/java/org/apache/hadoop/zebra/pig/TableStorer.java?rev=906244&r1=906243&r2=906244&view=diff
==============================================================================
--- hadoop/pig/branches/branch-0.6/contrib/zebra/src/java/org/apache/hadoop/zebra/pig/TableStorer.java (original)
+++ hadoop/pig/branches/branch-0.6/contrib/zebra/src/java/org/apache/hadoop/zebra/pig/TableStorer.java Wed Feb 3 21:08:18 2010
@@ -23,6 +23,8 @@
import java.lang.reflect.Constructor;
import java.util.List;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.BytesWritable;
@@ -43,6 +45,7 @@
import org.apache.pig.StoreConfig;
import org.apache.pig.CommittableStoreFunc;
import org.apache.pig.impl.logicalLayer.schema.Schema;
+import org.apache.pig.SortColInfo.Order;
import org.apache.pig.backend.hadoop.executionengine.util.MapRedUtil;
import org.apache.pig.data.Tuple;
@@ -128,19 +131,28 @@
{
org.apache.pig.SortColInfo sortColumn;
String sortColumnName;
+ boolean descending = false;
for (int i = 0; i < sortColumns.size(); i++)
{
sortColumn = sortColumns.get(i);
sortColumnName = sortColumn.getColName();
if (sortColumnName == null)
throw new IOException("Zebra does not support column positional reference yet");
+ if (sortColumn.getSortOrder() == Order.DESCENDING)
+ {
+ Log LOG = LogFactory.getLog(TableLoader.class);
+ LOG.warn("Sorting in descending order is not supported by Zebra and the table will be unsorted.");
+ descending = true;
+ break;
+ }
if (!org.apache.pig.data.DataType.isAtomic(schema.getField(sortColumnName).type))
throw new IOException(schema.getField(sortColumnName).alias+" is not of simple type as required for a sort column now.");
if (i > 0)
sb.append(",");
sb.append(sortColumnName);
}
- sortColumnNames = sb.toString();
+ if (!descending)
+ sortColumnNames = sb.toString();
}
}
try {
Added: hadoop/pig/branches/branch-0.6/contrib/zebra/src/test/org/apache/hadoop/zebra/pig/TestTableSortStorerDesc.java
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/branch-0.6/contrib/zebra/src/test/org/apache/hadoop/zebra/pig/TestTableSortStorerDesc.java?rev=906244&view=auto
==============================================================================
--- hadoop/pig/branches/branch-0.6/contrib/zebra/src/test/org/apache/hadoop/zebra/pig/TestTableSortStorerDesc.java (added)
+++ hadoop/pig/branches/branch-0.6/contrib/zebra/src/test/org/apache/hadoop/zebra/pig/TestTableSortStorerDesc.java Wed Feb 3 21:08:18 2010
@@ -0,0 +1,200 @@
+/**
+ * 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.hadoop.zebra.pig;
+
+import java.io.ByteArrayOutputStream;
+import java.io.File;
+import java.io.IOException;
+import java.io.PrintWriter;
+import java.util.Iterator;
+import java.util.StringTokenizer;
+
+import junit.framework.Assert;
+import junit.framework.TestCase;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.BytesWritable;
+import org.apache.hadoop.zebra.io.BasicTable;
+import org.apache.hadoop.zebra.io.TableInserter;
+import org.apache.hadoop.zebra.pig.TableStorer;
+import org.apache.hadoop.zebra.schema.Schema;
+import org.apache.hadoop.zebra.types.TypesUtils;
+import org.apache.pig.ExecType;
+import org.apache.pig.PigServer;
+import org.apache.pig.backend.executionengine.ExecException;
+import org.apache.pig.data.Tuple;
+import org.apache.pig.test.MiniCluster;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+/**
+ * Note:
+ *
+ * Make sure you add the build/pig-0.1.0-dev-core.jar to the Classpath of the
+ * app/debug configuration, when run this from inside the Eclipse.
+ *
+ */
+public class TestTableSortStorerDesc {
+ protected static ExecType execType = ExecType.MAPREDUCE;
+ private static MiniCluster cluster;
+ protected static PigServer pigServer;
+ private static Path pathTable;
+
+ @BeforeClass
+ public static void setUp() throws Exception {
+ if (System.getProperty("hadoop.log.dir") == null) {
+ String base = new File(".").getPath(); // getAbsolutePath();
+ System
+ .setProperty("hadoop.log.dir", new Path(base).toString() + "./logs");
+ }
+
+ if (execType == ExecType.MAPREDUCE) {
+ cluster = MiniCluster.buildCluster();
+ pigServer = new PigServer(ExecType.MAPREDUCE, cluster.getProperties());
+ } else {
+ pigServer = new PigServer(ExecType.LOCAL);
+ }
+
+ Configuration conf = new Configuration();
+ FileSystem fs = cluster.getFileSystem();
+ Path pathWorking = fs.getWorkingDirectory();
+ pathTable = new Path(pathWorking, "TestTableSortStorerDesc");
+ System.out.println("pathTable =" + pathTable);
+ BasicTable.Writer writer = new BasicTable.Writer(pathTable,
+ "SF_a:string,SF_b,SF_c,SF_d,SF_e,SF_f,SF_g",
+ "[SF_a, SF_b, SF_c]; [SF_e, SF_f, SF_g]", conf);
+ Schema schema = writer.getSchema();
+ Tuple tuple = TypesUtils.createTuple(schema);
+
+ final int numsBatch = 10;
+ final int numsInserters = 1;
+ TableInserter[] inserters = new TableInserter[numsInserters];
+ for (int i = 0; i < numsInserters; i++) {
+ inserters[i] = writer.getInserter("ins" + i, false);
+ }
+
+ for (int b = 0; b < numsBatch; b++) {
+ for (int i = 0; i < numsInserters; i++) {
+ TypesUtils.resetTuple(tuple);
+ for (int k = 0; k < tuple.size(); ++k) {
+ try {
+ tuple.set(k, (9-b) + "_" + i + "" + k);
+ } catch (ExecException e) {
+ e.printStackTrace();
+ }
+ }
+ inserters[i].insert(new BytesWritable(("key" + i).getBytes()), tuple);
+ }
+ }
+ for (int i = 0; i < numsInserters; i++) {
+ inserters[i].close();
+ }
+ writer.close();
+ }
+
+ @AfterClass
+ public static void tearDown() throws Exception {
+ pigServer.shutdown();
+ }
+
+ /**
+ * Return the name of the routine that called getCurrentMethodName
+ *
+ */
+ public String getCurrentMethodName() {
+ ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ PrintWriter pw = new PrintWriter(baos);
+ (new Throwable()).printStackTrace(pw);
+ pw.flush();
+ String stackTrace = baos.toString();
+ pw.close();
+
+ StringTokenizer tok = new StringTokenizer(stackTrace, "\n");
+ tok.nextToken(); // 'java.lang.Throwable'
+ tok.nextToken(); // 'at ...getCurrentMethodName'
+ String l = tok.nextToken(); // 'at ...<caller to getCurrentRoutine>'
+ // Parse line 3
+ tok = new StringTokenizer(l.trim(), " <(");
+ String t = tok.nextToken(); // 'at'
+ t = tok.nextToken(); // '...<caller to getCurrentRoutine>'
+ return t;
+ }
+
+ @Test
+ public void testStorer() throws ExecException, IOException {
+ /*
+ * Use pig LOAD to load testing data for store
+ */
+ String query = "records = LOAD '" + pathTable.toString()
+ + "' USING org.apache.hadoop.zebra.pig.TableLoader();";
+ pigServer.registerQuery(query);
+
+ Iterator<Tuple> it2 = pigServer.openIterator("records");
+ int row0 = 0;
+ Tuple RowValue2 = null;
+ while (it2.hasNext()) {
+ // Last row value
+ RowValue2 = it2.next();
+ row0++;
+ if (row0 == 10) {
+ Assert.assertEquals("0_01", RowValue2.get(1));
+ Assert.assertEquals("0_00", RowValue2.get(0).toString());
+ }
+ }
+ Assert.assertEquals(10, row0);
+
+ String orderby = "srecs = ORDER records BY SF_a DESC;";
+ pigServer.registerQuery(orderby);
+
+ /*
+ * Use pig STORE to store testing data BasicTable.Writer writer = new
+ * BasicTable.Writer(pathTable, "SF_a,SF_b,SF_c,SF_d,SF_e,SF_f,SF_g",
+ * "[SF_a, SF_b, SF_c]; [SF_e, SF_f, SF_g]", false, conf);
+ */
+ Path newPath = new Path(getCurrentMethodName());
+
+ pigServer
+ .store(
+ "srecs",
+ newPath.toString(),
+ TableStorer.class.getCanonicalName()
+ + "('[SF_a, SF_b, SF_c]; [SF_e]')");
+
+ // check new table content
+ String query3 = "newRecords = LOAD '"
+ + newPath.toString()
+ + "' USING org.apache.hadoop.zebra.pig.TableLoader('SF_a, SF_b');";
+ pigServer.registerQuery(query3);
+
+ Iterator<Tuple> it3 = pigServer.openIterator("newRecords");
+ int row = 0;
+ Tuple RowValue3 = null;
+ while (it3.hasNext()) {
+ // Last row value
+ RowValue3 = it3.next();
+ row++;
+ }
+ Assert.assertEquals(10, row);
+ }
+}