You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by al...@apache.org on 2015/06/08 21:43:40 UTC
[1/4] cassandra git commit: Remove deprecated legacy Hadoop code
Repository: cassandra
Updated Branches:
refs/heads/trunk d62cd1bc9 -> 446e25378
http://git-wip-us.apache.org/repos/asf/cassandra/blob/446e2537/test/pig/org/apache/cassandra/pig/ThriftColumnFamilyTest.java
----------------------------------------------------------------------
diff --git a/test/pig/org/apache/cassandra/pig/ThriftColumnFamilyTest.java b/test/pig/org/apache/cassandra/pig/ThriftColumnFamilyTest.java
index 60d04d3..93ac5be 100644
--- a/test/pig/org/apache/cassandra/pig/ThriftColumnFamilyTest.java
+++ b/test/pig/org/apache/cassandra/pig/ThriftColumnFamilyTest.java
@@ -22,24 +22,21 @@ import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Iterator;
+import com.datastax.driver.core.ResultSet;
+import com.datastax.driver.core.Row;
+import com.datastax.driver.core.Session;
import org.apache.cassandra.exceptions.ConfigurationException;
-import org.apache.cassandra.thrift.Cassandra;
-import org.apache.cassandra.thrift.ColumnOrSuperColumn;
-import org.apache.cassandra.thrift.ColumnPath;
-import org.apache.cassandra.thrift.ConsistencyLevel;
-import org.apache.cassandra.thrift.NotFoundException;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.pig.data.DataBag;
import org.apache.pig.data.DataByteArray;
import org.apache.pig.data.Tuple;
-import org.apache.thrift.TException;
import org.junit.Assert;
import org.junit.BeforeClass;
import org.junit.Test;
public class ThriftColumnFamilyTest extends PigTestBase
-{
+{
private static String[] statements = {
"DROP KEYSPACE IF EXISTS thrift_ks",
"CREATE KEYSPACE thrift_ks WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 1};",
@@ -171,7 +168,7 @@ public class ThriftColumnFamilyTest extends PigTestBase
};
@BeforeClass
- public static void setup() throws IOException, ConfigurationException, TException
+ public static void setup() throws IOException, ConfigurationException
{
startCassandra();
executeCQLStatements(statements);
@@ -300,11 +297,11 @@ public class ThriftColumnFamilyTest extends PigTestBase
}
@Test
- public void testCassandraStorageSchema() throws IOException
+ public void testCqlNativeStorageSchema() throws IOException
{
//results: (qux,(atomic_weight,0.660161815846869),(created,1335890877),(name,User Qux),(percent,64.7),
//(rating,2),(score,12000),(vote_type,dislike))
- pig.registerQuery("rows = LOAD 'cassandra://thrift_ks/some_app?" + defaultParameters + "' USING CassandraStorage();");
+ pig.registerQuery("rows = LOAD 'cql://thrift_ks/some_app?" + defaultParameters + "' USING CqlNativeStorage();");
//schema: {key: chararray,atomic_weight: (name: chararray,value: double),created: (name: chararray,value: long),
//name: (name: chararray,value: chararray),percent: (name: chararray,value: float),
@@ -339,12 +336,13 @@ public class ThriftColumnFamilyTest extends PigTestBase
}
@Test
- public void testCassandraStorageFullCopy() throws IOException, TException
+ public void testCqlNativeStorageFullCopy() throws IOException
{
pig.setBatchOn();
- pig.registerQuery("rows = LOAD 'cassandra://thrift_ks/some_app?" + defaultParameters + "' USING CassandraStorage();");
+ pig.registerQuery("rows = LOAD 'cql://thrift_ks/some_app?" + defaultParameters + nativeParameters + "&input_cql=select%20*%20from%20some_app%20where%20token(key)%20%3E%20%3F%20and%20token(key)%20%3C%3D%20%3F' USING CqlNativeStorage();");
+ pig.registerQuery("records = FOREACH rows GENERATE TOTUPLE(TOTUPLE('key', key)),TOTUPLE(atomic_weight, created, name, percent, rating, score, vote_type);");
//full copy
- pig.registerQuery("STORE rows INTO 'cassandra://thrift_ks/copy_of_some_app?" + defaultParameters + "' USING CassandraStorage();");
+ pig.registerQuery("STORE records INTO 'cql://thrift_ks/copy_of_some_app?" + defaultParameters + nativeParameters + "&output_query=UPDATE+thrift_ks.copy_of_some_app+set+atomic_weight+%3D+%3F,+created+%3D+%3F,+name+%3D+%3F,+percent+%3D+%3F,+rating+%3D+%3F,+score+%3D+%3F,+vote_type+%3D+%3F' USING CqlNativeStorage();");
pig.executeBatch();
Assert.assertEquals("User Qux", getColumnValue("thrift_ks", "copy_of_some_app", "name", "qux", "UTF8Type"));
Assert.assertEquals("dislike", getColumnValue("thrift_ks", "copy_of_some_app", "vote_type", "qux", "UTF8Type"));
@@ -352,158 +350,94 @@ public class ThriftColumnFamilyTest extends PigTestBase
}
@Test
- public void testCassandraStorageSingleTupleCopy() throws IOException, TException
+ public void testCqlNativeStorageSingleTupleCopy() throws IOException
{
executeCQLStatements(deleteCopyOfSomeAppTableData);
pig.setBatchOn();
- pig.registerQuery("rows = LOAD 'cassandra://thrift_ks/some_app?" + defaultParameters + "' USING CassandraStorage();");
+ pig.registerQuery("rows = LOAD 'cql://thrift_ks/some_app?" + defaultParameters + nativeParameters + "&input_cql=select%20*%20from%20some_app%20where%20token(key)%20%3E%20%3F%20and%20token(key)%20%3C%3D%20%3F' USING CqlNativeStorage();");
//single tuple
- pig.registerQuery("onecol = FOREACH rows GENERATE key, percent;");
- pig.registerQuery("STORE onecol INTO 'cassandra://thrift_ks/copy_of_some_app?" + defaultParameters + "' USING CassandraStorage();");
+ pig.registerQuery("onecol = FOREACH rows GENERATE TOTUPLE(TOTUPLE('key', key)), TOTUPLE(percent);");
+ pig.registerQuery("STORE onecol INTO 'cql://thrift_ks/copy_of_some_app?" + defaultParameters + nativeParameters + "&output_query=UPDATE+thrift_ks.copy_of_some_app+set+percent+%3D+%3F' USING CqlNativeStorage();");
pig.executeBatch();
- String value = null;
- try
- {
- value = getColumnValue("thrift_ks", "copy_of_some_app", "name", "qux", "UTF8Type");
- }
- catch (NotFoundException e)
- {
- Assert.assertTrue(true);
- }
+ String value = getColumnValue("thrift_ks", "copy_of_some_app", "name", "qux", "UTF8Type");
if (value != null)
Assert.fail();
- try
- {
- value = getColumnValue("thrift_ks", "copy_of_some_app", "vote_type", "qux", "UTF8Type");
- }
- catch (NotFoundException e)
- {
- Assert.assertTrue(true);
- }
+ value = getColumnValue("thrift_ks", "copy_of_some_app", "vote_type", "qux", "UTF8Type");
if (value != null)
Assert.fail();
Assert.assertEquals("64.7", getColumnValue("thrift_ks", "copy_of_some_app", "percent", "qux", "FloatType"));
}
@Test
- public void testCassandraStorageBagOnlyCopy() throws IOException, TException
+ public void testCqlNativeStorageBagOnlyCopy() throws IOException
{
executeCQLStatements(deleteCopyOfSomeAppTableData);
pig.setBatchOn();
- pig.registerQuery("rows = LOAD 'cassandra://thrift_ks/some_app?" + defaultParameters + "' USING CassandraStorage();");
+ pig.registerQuery("rows = LOAD 'cql://thrift_ks/some_app?" + defaultParameters + nativeParameters + "&input_cql=select%20*%20from%20some_app%20where%20token(key)%20%3E%20%3F%20and%20token(key)%20%3C%3D%20%3F' USING CqlNativeStorage();");
//bag only
- pig.registerQuery("other = FOREACH rows GENERATE key, columns;");
- pig.registerQuery("STORE other INTO 'cassandra://thrift_ks/copy_of_some_app?" + defaultParameters + "' USING CassandraStorage();");
+ pig.registerQuery("other = FOREACH rows GENERATE TOTUPLE(TOTUPLE('key', key)), TOTUPLE();");
+ pig.registerQuery("STORE other INTO 'cql://thrift_ks/copy_of_some_app?" + defaultParameters + nativeParameters + "' USING CqlNativeStorage();");
pig.executeBatch();
- String value = null;
- try
- {
- value = getColumnValue("thrift_ks", "copy_of_some_app", "name", "qux", "UTF8Type");
- }
- catch (NotFoundException e)
- {
- Assert.assertTrue(true);
- }
+ String value = getColumnValue("thrift_ks", "copy_of_some_app", "name", "qux", "UTF8Type");
if (value != null)
Assert.fail();
- try
- {
- value = getColumnValue("thrift_ks", "copy_of_some_app", "vote_type", "qux", "UTF8Type");
- }
- catch (NotFoundException e)
- {
- Assert.assertTrue(true);
- }
+ value = getColumnValue("thrift_ks", "copy_of_some_app", "vote_type", "qux", "UTF8Type");
if (value != null)
Assert.fail();
- try
- {
- value = getColumnValue("thrift_ks", "copy_of_some_app", "percent", "qux", "FloatType");
- }
- catch (NotFoundException e)
- {
- Assert.assertTrue(true);
- }
+ value = getColumnValue("thrift_ks", "copy_of_some_app", "percent", "qux", "FloatType");
if (value != null)
Assert.fail();
}
@Test
- public void testCassandraStorageFilter() throws IOException, TException
+ public void testCqlNativeStorageFilter() throws IOException
{
executeCQLStatements(deleteCopyOfSomeAppTableData);
pig.setBatchOn();
- pig.registerQuery("rows = LOAD 'cassandra://thrift_ks/some_app?" + defaultParameters + "' USING CassandraStorage();");
+ pig.registerQuery("rows = LOAD 'cql://thrift_ks/some_app?" + defaultParameters + nativeParameters + "&input_cql=select%20*%20from%20some_app%20where%20token(key)%20%3E%20%3F%20and%20token(key)%20%3C%3D%20%3F' USING CqlNativeStorage();");
//filter
- pig.registerQuery("likes = FILTER rows by vote_type.value eq 'like' and rating.value > 5;");
- pig.registerQuery("STORE likes INTO 'cassandra://thrift_ks/copy_of_some_app?" + defaultParameters + "' USING CassandraStorage();");
+ pig.registerQuery("likes = FILTER rows by vote_type eq 'like' and rating > 5;");
+ pig.registerQuery("records = FOREACH likes GENERATE TOTUPLE(TOTUPLE('key', key)),TOTUPLE(atomic_weight, created, name, percent, rating, score, vote_type);");
+ pig.registerQuery("STORE records INTO 'cql://thrift_ks/copy_of_some_app?" + defaultParameters + nativeParameters + "&output_query=UPDATE+thrift_ks.copy_of_some_app+set+atomic_weight+%3D+%3F,+created+%3D+%3F,+name+%3D+%3F,+percent+%3D+%3F,+rating+%3D+%3F,+score+%3D+%3F,+vote_type+%3D+%3F' USING CqlNativeStorage();");
pig.executeBatch();
Assert.assertEquals("like", getColumnValue("thrift_ks", "copy_of_some_app", "vote_type", "bar", "UTF8Type"));
Assert.assertEquals("like", getColumnValue("thrift_ks", "copy_of_some_app", "vote_type", "foo", "UTF8Type"));
- String value = null;
- try
- {
- value = getColumnValue("thrift_ks", "copy_of_some_app", "vote_type", "qux", "UTF8Type");
- }
- catch (NotFoundException e)
- {
- Assert.assertTrue(true);
- }
+ String value = getColumnValue("thrift_ks", "copy_of_some_app", "vote_type", "qux", "UTF8Type");
if (value != null)
Assert.fail();
- try
- {
- value = getColumnValue("thrift_ks", "copy_of_some_app", "vote_type", "baz", "UTF8Type");
- }
- catch (NotFoundException e)
- {
- Assert.assertTrue(true);
- }
+ value = getColumnValue("thrift_ks", "copy_of_some_app", "vote_type", "baz", "UTF8Type");
+
if (value != null)
Assert.fail();
executeCQLStatements(deleteCopyOfSomeAppTableData);
pig.setBatchOn();
- pig.registerQuery("rows = LOAD 'cassandra://thrift_ks/some_app?" + defaultParameters + "' USING CassandraStorage();");
- pig.registerQuery("dislikes_extras = FILTER rows by vote_type.value eq 'dislike';");
- pig.registerQuery("STORE dislikes_extras INTO 'cassandra://thrift_ks/copy_of_some_app?" + defaultParameters + "' USING CassandraStorage();");
+ pig.registerQuery("rows = LOAD 'cql://thrift_ks/some_app?" + defaultParameters + nativeParameters + "&input_cql=select%20*%20from%20some_app%20where%20token(key)%20%3E%20%3F%20and%20token(key)%20%3C%3D%20%3F' USING CqlNativeStorage();");
+ pig.registerQuery("dislikes_extras = FILTER rows by vote_type eq 'dislike';");
+ pig.registerQuery("dislikes_records = FOREACH dislikes_extras GENERATE TOTUPLE(TOTUPLE('key', key)),TOTUPLE(atomic_weight, created, name, percent, rating, score, vote_type);");
+ pig.registerQuery("STORE dislikes_records INTO 'cql://thrift_ks/copy_of_some_app?" + defaultParameters + nativeParameters + "&output_query=UPDATE+thrift_ks.copy_of_some_app+set+atomic_weight+%3D+%3F,+created+%3D+%3F,+name+%3D+%3F,+percent+%3D+%3F,+rating+%3D+%3F,+score+%3D+%3F,+vote_type+%3D+%3F' USING CqlNativeStorage();");
pig.executeBatch();
Assert.assertEquals("dislike", getColumnValue("thrift_ks", "copy_of_some_app", "vote_type", "baz", "UTF8Type"));
Assert.assertEquals("dislike", getColumnValue("thrift_ks", "copy_of_some_app", "vote_type", "qux", "UTF8Type"));
- value = null;
- try
- {
- value = getColumnValue("thrift_ks", "copy_of_some_app", "vote_type", "bar", "UTF8Type");
- }
- catch (NotFoundException e)
- {
- Assert.assertTrue(true);
- }
+ value = getColumnValue("thrift_ks", "copy_of_some_app", "vote_type", "bar", "UTF8Type");
if (value != null)
Assert.fail();
- try
- {
- value = getColumnValue("thrift_ks", "copy_of_some_app", "vote_type", "foo", "UTF8Type");
- }
- catch (NotFoundException e)
- {
- Assert.assertTrue(true);
- }
+ value = getColumnValue("thrift_ks", "copy_of_some_app", "vote_type", "foo", "UTF8Type");
if (value != null)
Assert.fail();
}
@Test
- public void testCassandraStorageJoin() throws IOException
+ public void testCqlNativeStorageJoin() throws IOException
{
//test key types with a join
- pig.registerQuery("U8 = load 'cassandra://thrift_ks/u8?" + defaultParameters + "' using CassandraStorage();");
- pig.registerQuery("Bytes = load 'cassandra://thrift_ks/bytes?" + defaultParameters + "' using CassandraStorage();");
+ pig.registerQuery("U8 = load 'cql://thrift_ks/u8?" + defaultParameters + nativeParameters + "&input_cql=select%20*%20from%20u8%20where%20token(key)%20%3E%20%3F%20and%20token(key)%20%3C%3D%20%3F' using CqlNativeStorage();");
+ pig.registerQuery("Bytes = load 'cql://thrift_ks/bytes?" + defaultParameters + nativeParameters + "&input_cql=select%20*%20from%20bytes%20where%20token(key)%20%3E%20%3F%20and%20token(key)%20%3C%3D%20%3F' using CqlNativeStorage();");
//cast key to chararray
- pig.registerQuery("b = foreach Bytes generate (chararray)key, columns;");
+ pig.registerQuery("b = foreach Bytes generate (chararray)key, column1, value;");
//key in Bytes is a bytearray, U8 chararray
//(foo,{(x,Z)},foo,{(x,Z)})
@@ -512,18 +446,11 @@ public class ThriftColumnFamilyTest extends PigTestBase
if (it.hasNext()) {
Tuple t = it.next();
Assert.assertEquals(t.get(0), new DataByteArray("foo".getBytes()));
- DataBag columns = (DataBag) t.get(1);
- Iterator<Tuple> iter = columns.iterator();
- Tuple t1 = iter.next();
- Assert.assertEquals(t1.get(0), "x");
- Assert.assertEquals(t1.get(1), new DataByteArray("Z".getBytes()));
- String column = (String) t.get(2);
- Assert.assertEquals(column, "foo");
- columns = (DataBag) t.get(3);
- iter = columns.iterator();
- Tuple t2 = iter.next();
- Assert.assertEquals(t2.get(0), "x");
- Assert.assertEquals(t2.get(1), new DataByteArray("Z".getBytes()));
+ Assert.assertEquals(t.get(1), "x");
+ Assert.assertEquals(t.get(2), new DataByteArray("Z".getBytes()));
+ Assert.assertEquals(t.get(3), "foo");
+ Assert.assertEquals(t.get(4), "x");
+ Assert.assertEquals(t.get(5), new DataByteArray("Z".getBytes()));
}
//key should now be cast into a chararray
//(foo,{(x,Z)},foo,{(x,Z)})
@@ -532,27 +459,22 @@ public class ThriftColumnFamilyTest extends PigTestBase
if (it.hasNext()) {
Tuple t = it.next();
Assert.assertEquals(t.get(0), "foo");
- DataBag columns = (DataBag) t.get(1);
- Iterator<Tuple> iter = columns.iterator();
- Tuple t1 = iter.next();
- Assert.assertEquals(t1.get(0), "x");
- Assert.assertEquals(t1.get(1), new DataByteArray("Z".getBytes()));
- String column = (String) t.get(2);
- Assert.assertEquals(column, "foo");
- columns = (DataBag) t.get(3);
- iter = columns.iterator();
- Tuple t2 = iter.next();
- Assert.assertEquals(t2.get(0), "x");
- Assert.assertEquals(t2.get(1), new DataByteArray("Z".getBytes()));
+ Assert.assertEquals(t.get(1), "x");
+ Assert.assertEquals(t.get(2), new DataByteArray("Z".getBytes()));
+ Assert.assertEquals(t.get(3), "foo");
+ Assert.assertEquals(t.get(4), "x");
+ Assert.assertEquals(t.get(5), new DataByteArray("Z".getBytes()));
}
}
@Test
- public void testCassandraStorageCounterCF() throws IOException
+ public void testCqlNativeStorageCounterCF() throws IOException
{
//Test counter column family support
- pig.registerQuery("CC = load 'cassandra://thrift_ks/cc?" + defaultParameters + "' using CassandraStorage();");
- pig.registerQuery("total_hits = foreach CC generate key, SUM(columns.value);");
+ pig.registerQuery("CC = load 'cql://thrift_ks/cc?" + defaultParameters + nativeParameters + "&input_cql=select%20*%20from%20cc%20where%20token(key)%20%3E%20%3F%20and%20token(key)%20%3C%3D%20%3F' using CqlNativeStorage();");
+ pig.registerQuery("A = foreach CC generate key, name, value;");
+ pig.registerQuery("B = GROUP A BY key;");
+ pig.registerQuery("total_hits = foreach B generate group, SUM(A.value);");
//(chuck,4)
Tuple t = pig.openIterator("total_hits").next();
Assert.assertEquals(t.get(0), "chuck");
@@ -560,12 +482,11 @@ public class ThriftColumnFamilyTest extends PigTestBase
}
@Test
- public void testCassandraStorageCompositeColumnCF() throws IOException
+ public void testCqlNativeStorageCompositeColumnCF() throws IOException
{
//Test CompositeType
- pig.registerQuery("compo = load 'cassandra://thrift_ks/compo?" + defaultParameters + "' using CassandraStorage();");
- pig.registerQuery("compo = foreach compo generate key as method, flatten(columns);");
- pig.registerQuery("lee = filter compo by columns::name == ('bruce','lee');");
+ pig.registerQuery("compo = load 'cql://thrift_ks/compo?" + defaultParameters + nativeParameters + "&input_cql=select%20*%20from%20compo%20where%20token(key)%20%3E%20%3F%20and%20token(key)%20%3C%3D%20%3F' using CqlNativeStorage();");
+ pig.registerQuery("lee = filter compo by column1 == 'bruce' AND column2 == 'lee';");
//(kick,(bruce,lee),oww)
//(punch,(bruce,lee),ouch)
@@ -574,18 +495,16 @@ public class ThriftColumnFamilyTest extends PigTestBase
while (it.hasNext()) {
count ++;
Tuple t = it.next();
- Tuple t1 = (Tuple) t.get(1);
- Assert.assertEquals(t1.get(0), "bruce");
- Assert.assertEquals(t1.get(1), "lee");
+ Assert.assertEquals(t.get(1), "bruce");
+ Assert.assertEquals(t.get(2), "lee");
if ("kick".equals(t.get(0)))
- Assert.assertEquals(t.get(2), "oww");
- else if ("kick".equals(t.get(0)))
- Assert.assertEquals(t.get(2), "ouch");
+ Assert.assertEquals(t.get(3), "oww");
+ else
+ Assert.assertEquals(t.get(3), "ouch");
}
Assert.assertEquals(count, 2);
- pig.registerQuery("night = load 'cassandra://thrift_ks/compo_int?" + defaultParameters + "' using CassandraStorage();");
- pig.registerQuery("night = foreach night generate flatten(columns);");
- pig.registerQuery("night = foreach night generate (int)columns::name.$0+(double)columns::name.$1/60 as hour, columns::value as noise;");
+ pig.registerQuery("night = load 'cql://thrift_ks/compo_int?" + defaultParameters + nativeParameters + "&input_cql=select%20*%20from%20compo_int%20where%20token(key)%20%3E%20%3F%20and%20token(key)%20%3C%3D%20%3F' using CqlNativeStorage();");
+ pig.registerQuery("night = foreach night generate (int)column1+(double)column2/60 as hour, value as noise;");
//What happens at the darkest hour?
pig.registerQuery("darkest = filter night by hour > 2 and hour < 5;");
@@ -598,10 +517,10 @@ public class ThriftColumnFamilyTest extends PigTestBase
Assert.assertEquals(t.get(1), "daddy?");
}
pig.setBatchOn();
- pig.registerQuery("compo_int_rows = LOAD 'cassandra://thrift_ks/compo_int?" + defaultParameters + "' using CassandraStorage();");
- pig.registerQuery("STORE compo_int_rows INTO 'cassandra://thrift_ks/compo_int_copy?" + defaultParameters + "' using CassandraStorage();");
+ pig.registerQuery("compo_int_rows = LOAD 'cql://thrift_ks/compo_int?" + defaultParameters + nativeParameters + "&input_cql=select%20*%20from%20compo_int%20where%20token(key)%20%3E%20%3F%20and%20token(key)%20%3C%3D%20%3F' using CqlNativeStorage();");
+ pig.registerQuery("STORE compo_int_rows INTO 'cql://thrift_ks/compo_int_copy?" + defaultParameters + nativeParameters + "&output_query=UPDATE+thrift_ks.compo_int_copy+set+column1+%3D+%3F,+column2+%3D+%3F,+value+%3D+%3F' using CqlNativeStorage();");
pig.executeBatch();
- pig.registerQuery("compocopy_int_rows = LOAD 'cassandra://thrift_ks/compo_int_copy?" + defaultParameters + "' using CassandraStorage();");
+ pig.registerQuery("compocopy_int_rows = LOAD 'cql://thrift_ks/compo_int_copy?" + defaultParameters + nativeParameters + "&input_cql=select%20*%20from%20compo_int_copy%20where%20token(key)%20%3E%20%3F%20and%20token(key)%20%3C%3D%20%3F' using CqlNativeStorage();");
//(clock,{((1,0),z),((1,30),zzzz),((2,30),daddy?),((6,30),coffee...)})
it = pig.openIterator("compocopy_int_rows");
count = 0;
@@ -627,32 +546,26 @@ public class ThriftColumnFamilyTest extends PigTestBase
}
@Test
- public void testCassandraStorageCompositeKeyCF() throws IOException
+ public void testCqlNativeStorageCompositeKeyCF() throws IOException
{
//Test CompositeKey
- pig.registerQuery("compokeys = load 'cassandra://thrift_ks/compo_key?" + defaultParameters + "' using CassandraStorage();");
- pig.registerQuery("compokeys = filter compokeys by key.$1 == 40;");
- //((clock,40),{(6,coffee...)})
+ pig.registerQuery("compokeys = load 'cql://thrift_ks/compo_key?" + defaultParameters + nativeParameters + "&input_cql=select%20*%20from%20compo_key%20where%20token(key,column1)%20%3E%20%3F%20and%20token(key,column1)%20%3C%3D%20%3F' using CqlNativeStorage();");
+ pig.registerQuery("compokeys = filter compokeys by column1 == 40;");
+ //(clock,40,6,coffee...)
Iterator<Tuple> it = pig.openIterator("compokeys");
if (it.hasNext()) {
Tuple t = it.next();
- Tuple key = (Tuple) t.get(0);
- Assert.assertEquals(key.get(0), "clock");
- Assert.assertEquals(key.get(1), 40L);
- DataBag columns = (DataBag) t.get(1);
- Iterator<Tuple> iter = columns.iterator();
- if (iter.hasNext())
- {
- Tuple t1 = iter.next();
- Assert.assertEquals(t1.get(0), 6L);
- Assert.assertEquals(t1.get(1), "coffee...");
- }
+ Assert.assertEquals(t.get(0), "clock");
+ Assert.assertEquals(t.get(1), 40L);
+ Assert.assertEquals(t.get(2), 6L);
+ Assert.assertEquals(t.get(3), "coffee...");
}
pig.setBatchOn();
- pig.registerQuery("compo_key_rows = LOAD 'cassandra://thrift_ks/compo_key?" + defaultParameters + "' using CassandraStorage();");
- pig.registerQuery("STORE compo_key_rows INTO 'cassandra://thrift_ks/compo_key_copy?" + defaultParameters + "' using CassandraStorage();");
+ pig.registerQuery("compo_key_rows = LOAD 'cql://thrift_ks/compo_key?" + defaultParameters + nativeParameters + "&input_cql=select%20*%20from%20compo_key%20where%20token(key,column1)%20%3E%20%3F%20and%20token(key,column1)%20%3C%3D%20%3F' using CqlNativeStorage();");
+ pig.registerQuery("compo_key_rows = FOREACH compo_key_rows GENERATE TOTUPLE(TOTUPLE('key',key),TOTUPLE('column1',column1),TOTUPLE('column2',column2)),TOTUPLE(value);");
+ pig.registerQuery("STORE compo_key_rows INTO 'cql://thrift_ks/compo_key_copy?" + defaultParameters + nativeParameters + "&output_query=UPDATE+thrift_ks.compo_key_copy+set+value+%3D+%3F' using CqlNativeStorage();");
pig.executeBatch();
- pig.registerQuery("compo_key_copy_rows = LOAD 'cassandra://thrift_ks/compo_key_copy?" + defaultParameters + "' using CassandraStorage();");
+ pig.registerQuery("compo_key_copy_rows = LOAD 'cql://thrift_ks/compo_key_copy?" + defaultParameters + nativeParameters + "&input_cql=select%20*%20from%20compo_key_copy%20where%20token(key,column1)%20%3E%20%3F%20and%20token(key,column1)%20%3C%3D%20%3F' using CqlNativeStorage();");
//((clock,10),{(1,z)})
//((clock,20),{(1,zzzz)})
//((clock,30),{(2,daddy?)})
@@ -662,66 +575,43 @@ public class ThriftColumnFamilyTest extends PigTestBase
while (it.hasNext()) {
Tuple t = it.next();
count ++;
- Tuple key = (Tuple) t.get(0);
- if ("clock".equals(key.get(0)) && (Long) key.get(1) == 10L)
+ if ("clock".equals(t.get(0)) && (Long) t.get(1) == 10L)
{
- DataBag columns = (DataBag) t.get(1);
- Iterator<Tuple> iter = columns.iterator();
- if (iter.hasNext())
- {
- Tuple t1 = iter.next();
- Assert.assertEquals(t1.get(0), 1L);
- Assert.assertEquals(t1.get(1), "z");
- }
+ Assert.assertEquals(t.get(2), 1L);
+ Assert.assertEquals(t.get(3), "z");
}
- else if ("clock".equals(key.get(0)) && (Long) key.get(1) == 40L)
+ else if ("clock".equals(t.get(0)) && (Long) t.get(1) == 40L)
{
- DataBag columns = (DataBag) t.get(1);
- Iterator<Tuple> iter = columns.iterator();
- if (iter.hasNext())
- {
- Tuple t1 = iter.next();
- Assert.assertEquals(t1.get(0), 6L);
- Assert.assertEquals(t1.get(1), "coffee...");
- }
+ Assert.assertEquals(t.get(2), 6L);
+ Assert.assertEquals(t.get(3), "coffee...");
}
- else if ("clock".equals(key.get(0)) && (Long) key.get(1) == 20L)
+ else if ("clock".equals(t.get(0)) && (Long) t.get(1) == 20L)
{
- DataBag columns = (DataBag) t.get(1);
- Iterator<Tuple> iter = columns.iterator();
- if (iter.hasNext())
- {
- Tuple t1 = iter.next();
- Assert.assertEquals(t1.get(0), 1L);
- Assert.assertEquals(t1.get(1), "zzzz");
- }
+ Assert.assertEquals(t.get(2), 1L);
+ Assert.assertEquals(t.get(3), "zzzz");
}
- else if ("clock".equals(key.get(0)) && (Long) key.get(1) == 30L)
+ else if ("clock".equals(t.get(0)) && (Long) t.get(1) == 30L)
{
- DataBag columns = (DataBag) t.get(1);
- Iterator<Tuple> iter = columns.iterator();
- if (iter.hasNext())
- {
- Tuple t1 = iter.next();
- Assert.assertEquals(t1.get(0), 2L);
- Assert.assertEquals(t1.get(1), "daddy?");
- }
+ Assert.assertEquals(t.get(2), 2L);
+ Assert.assertEquals(t.get(3), "daddy?");
}
}
Assert.assertEquals(4, count);
}
- private String getColumnValue(String ks, String cf, String colName, String key, String validator) throws TException, IOException
+ private String getColumnValue(String ks, String cf, String colName, String key, String validator) throws IOException
{
- Cassandra.Client client = getClient();
- client.set_keyspace(ks);
+ Session client = getClient();
+ client.execute("USE " + ks);
+
+ String query = String.format("SELECT %s FROM %s WHERE key = '%s'", colName, cf, key);
+
+ ResultSet rows = client.execute(query);
+ Row row = rows.one();
- ByteBuffer key_user_id = ByteBufferUtil.bytes(key);
- ColumnPath cp = new ColumnPath(cf);
- cp.column = ByteBufferUtil.bytes(colName);
+ if (row == null || row.isNull(0))
+ return null;
- // read
- ColumnOrSuperColumn got = client.get(key_user_id, cp, ConsistencyLevel.ONE);
- return parseType(validator).getString(got.getColumn().value);
+ return parseType(validator).getString(row.getBytesUnsafe(0));
}
}
[3/4] cassandra git commit: Remove deprecated legacy Hadoop code
Posted by al...@apache.org.
http://git-wip-us.apache.org/repos/asf/cassandra/blob/446e2537/src/java/org/apache/cassandra/hadoop/ColumnFamilyOutputFormat.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/hadoop/ColumnFamilyOutputFormat.java b/src/java/org/apache/cassandra/hadoop/ColumnFamilyOutputFormat.java
deleted file mode 100644
index 92e3829..0000000
--- a/src/java/org/apache/cassandra/hadoop/ColumnFamilyOutputFormat.java
+++ /dev/null
@@ -1,183 +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.cassandra.hadoop;
-
-
-import java.io.*;
-import java.nio.ByteBuffer;
-import java.util.*;
-
-import org.slf4j.*;
-
-import org.apache.cassandra.auth.*;
-import org.apache.cassandra.thrift.*;
-import org.apache.hadoop.conf.*;
-import org.apache.hadoop.mapreduce.*;
-import org.apache.thrift.protocol.*;
-import org.apache.thrift.transport.*;
-
-/**
- * The <code>ColumnFamilyOutputFormat</code> acts as a Hadoop-specific
- * OutputFormat that allows reduce tasks to store keys (and corresponding
- * values) as Cassandra rows (and respective columns) in a given
- * ColumnFamily.
- *
- * <p>
- * As is the case with the {@link ColumnFamilyInputFormat}, you need to set the
- * Keyspace and ColumnFamily in your
- * Hadoop job Configuration. The {@link ConfigHelper} class, through its
- * {@link ConfigHelper#setOutputColumnFamily} method, is provided to make this
- * simple.
- * </p>
- *
- * <p>
- * For the sake of performance, this class employs a lazy write-back caching
- * mechanism, where its record writer batches mutations created based on the
- * reduce's inputs (in a task-specific map), and periodically makes the changes
- * official by sending a batch mutate request to Cassandra.
- * </p>
- */
-@Deprecated
-public class ColumnFamilyOutputFormat extends OutputFormat<ByteBuffer,List<Mutation>>
- implements org.apache.hadoop.mapred.OutputFormat<ByteBuffer,List<Mutation>>
-{
- public static final String BATCH_THRESHOLD = "mapreduce.output.columnfamilyoutputformat.batch.threshold";
- public static final String QUEUE_SIZE = "mapreduce.output.columnfamilyoutputformat.queue.size";
-
- private static final Logger logger = LoggerFactory.getLogger(ColumnFamilyOutputFormat.class);
-
- /**
- * The OutputCommitter for this format does not write any data to the DFS.
- *
- * @param context
- * the task context
- * @return an output committer
- * @throws IOException
- * @throws InterruptedException
- */
- public OutputCommitter getOutputCommitter(TaskAttemptContext context) throws IOException, InterruptedException
- {
- return new NullOutputCommitter();
- }
-
- /**
- * Check for validity of the output-specification for the job.
- *
- * @param context
- * information about the job
- */
- public void checkOutputSpecs(JobContext context)
- {
- checkOutputSpecs(HadoopCompat.getConfiguration(context));
- }
-
- protected void checkOutputSpecs(Configuration conf)
- {
- if (ConfigHelper.getOutputKeyspace(conf) == null)
- throw new UnsupportedOperationException("You must set the keyspace with setOutputKeyspace()");
- if (ConfigHelper.getOutputPartitioner(conf) == null)
- throw new UnsupportedOperationException("You must set the output partitioner to the one used by your Cassandra cluster");
- if (ConfigHelper.getOutputInitialAddress(conf) == null)
- throw new UnsupportedOperationException("You must set the initial output address to a Cassandra node");
- }
-
- /** Fills the deprecated OutputFormat interface for streaming. */
- @Deprecated
- public void checkOutputSpecs(org.apache.hadoop.fs.FileSystem filesystem, org.apache.hadoop.mapred.JobConf job) throws IOException
- {
- checkOutputSpecs(job);
- }
-
- /**
- * Connects to the given server:port and returns a client based on the given socket that points to the configured
- * keyspace, and is logged in with the configured credentials.
- *
- * @param host fully qualified host name to connect to
- * @param port RPC port of the server
- * @param conf a job configuration
- * @return a cassandra client
- * @throws Exception set of thrown exceptions may be implementation defined,
- * depending on the used transport factory
- */
- @SuppressWarnings("resource")
- public static Cassandra.Client createAuthenticatedClient(String host, int port, Configuration conf) throws Exception
- {
- logger.debug("Creating authenticated client for CF output format");
- TTransport transport = ConfigHelper.getClientTransportFactory(conf).openTransport(host, port);
- TProtocol binaryProtocol = new TBinaryProtocol(transport, true, true);
- Cassandra.Client client = new Cassandra.Client(binaryProtocol);
- client.set_keyspace(ConfigHelper.getOutputKeyspace(conf));
- String user = ConfigHelper.getOutputKeyspaceUserName(conf);
- String password = ConfigHelper.getOutputKeyspacePassword(conf);
- if ((user != null) && (password != null))
- login(user, password, client);
-
- logger.debug("Authenticated client for CF output format created successfully");
- return client;
- }
-
- public static void login(String user, String password, Cassandra.Client client) throws Exception
- {
- Map<String, String> creds = new HashMap<String, String>();
- creds.put(PasswordAuthenticator.USERNAME_KEY, user);
- creds.put(PasswordAuthenticator.PASSWORD_KEY, password);
- AuthenticationRequest authRequest = new AuthenticationRequest(creds);
- client.login(authRequest);
- }
-
- /** Fills the deprecated OutputFormat interface for streaming. */
- @Deprecated
- public ColumnFamilyRecordWriter getRecordWriter(org.apache.hadoop.fs.FileSystem filesystem, org.apache.hadoop.mapred.JobConf job, String name, org.apache.hadoop.util.Progressable progress)
- {
- return new ColumnFamilyRecordWriter(job, progress);
- }
-
- /**
- * Get the {@link RecordWriter} for the given task.
- *
- * @param context
- * the information about the current task.
- * @return a {@link RecordWriter} to write the output for the job.
- */
- public ColumnFamilyRecordWriter getRecordWriter(final TaskAttemptContext context) throws InterruptedException
- {
- return new ColumnFamilyRecordWriter(context);
- }
-
- /**
- * An {@link OutputCommitter} that does nothing.
- */
- private static class NullOutputCommitter extends OutputCommitter
- {
- public void abortTask(TaskAttemptContext taskContext) { }
-
- public void cleanupJob(JobContext jobContext) { }
-
- public void commitTask(TaskAttemptContext taskContext) { }
-
- public boolean needsTaskCommit(TaskAttemptContext taskContext)
- {
- return false;
- }
-
- public void setupJob(JobContext jobContext) { }
-
- public void setupTask(TaskAttemptContext taskContext) { }
- }
-
-}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/446e2537/src/java/org/apache/cassandra/hadoop/ColumnFamilyRecordReader.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/hadoop/ColumnFamilyRecordReader.java b/src/java/org/apache/cassandra/hadoop/ColumnFamilyRecordReader.java
deleted file mode 100644
index aee730d..0000000
--- a/src/java/org/apache/cassandra/hadoop/ColumnFamilyRecordReader.java
+++ /dev/null
@@ -1,615 +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.cassandra.hadoop;
-
-import java.io.IOException;
-import java.net.InetAddress;
-import java.net.UnknownHostException;
-import java.nio.ByteBuffer;
-import java.util.*;
-
-import com.google.common.collect.*;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import org.apache.cassandra.schema.LegacySchemaTables;
-import org.apache.cassandra.db.SystemKeyspace;
-import org.apache.cassandra.db.marshal.AbstractType;
-import org.apache.cassandra.db.marshal.CompositeType;
-import org.apache.cassandra.db.marshal.TypeParser;
-import org.apache.cassandra.dht.IPartitioner;
-import org.apache.cassandra.exceptions.ConfigurationException;
-import org.apache.cassandra.thrift.*;
-import org.apache.cassandra.utils.ByteBufferUtil;
-import org.apache.cassandra.utils.FBUtilities;
-import org.apache.cassandra.utils.Pair;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.mapreduce.InputSplit;
-import org.apache.hadoop.mapreduce.RecordReader;
-import org.apache.hadoop.mapreduce.TaskAttemptContext;
-import org.apache.thrift.TException;
-import org.apache.thrift.transport.TTransport;
-
-@Deprecated
-public class ColumnFamilyRecordReader extends RecordReader<ByteBuffer, SortedMap<ByteBuffer, ColumnFamilyRecordReader.Column>>
- implements org.apache.hadoop.mapred.RecordReader<ByteBuffer, SortedMap<ByteBuffer, ColumnFamilyRecordReader.Column>>
-{
- private static final Logger logger = LoggerFactory.getLogger(ColumnFamilyRecordReader.class);
-
- public static final int CASSANDRA_HADOOP_MAX_KEY_SIZE_DEFAULT = 8192;
-
- private ColumnFamilySplit split;
- private RowIterator iter;
- private Pair<ByteBuffer, SortedMap<ByteBuffer, Column>> currentRow;
- private SlicePredicate predicate;
- private boolean isEmptyPredicate;
- private int totalRowCount; // total number of rows to fetch
- private int batchSize; // fetch this many per batch
- private String keyspace;
- private String cfName;
- private Cassandra.Client client;
- private ConsistencyLevel consistencyLevel;
- private int keyBufferSize = 8192;
- private List<IndexExpression> filter;
-
-
- public ColumnFamilyRecordReader()
- {
- this(ColumnFamilyRecordReader.CASSANDRA_HADOOP_MAX_KEY_SIZE_DEFAULT);
- }
-
- public ColumnFamilyRecordReader(int keyBufferSize)
- {
- super();
- this.keyBufferSize = keyBufferSize;
- }
-
- @SuppressWarnings("resource")
- public void close()
- {
- if (client != null)
- {
- TTransport transport = client.getOutputProtocol().getTransport();
- if (transport.isOpen())
- transport.close();
- }
- }
-
- public ByteBuffer getCurrentKey()
- {
- return currentRow.left;
- }
-
- public SortedMap<ByteBuffer, Column> getCurrentValue()
- {
- return currentRow.right;
- }
-
- public float getProgress()
- {
- if (!iter.hasNext())
- return 1.0F;
-
- // the progress is likely to be reported slightly off the actual but close enough
- float progress = ((float) iter.rowsRead() / totalRowCount);
- return progress > 1.0F ? 1.0F : progress;
- }
-
- static boolean isEmptyPredicate(SlicePredicate predicate)
- {
- if (predicate == null)
- return true;
-
- if (predicate.isSetColumn_names() && predicate.getSlice_range() == null)
- return false;
-
- if (predicate.getSlice_range() == null)
- return true;
-
- byte[] start = predicate.getSlice_range().getStart();
- if ((start != null) && (start.length > 0))
- return false;
-
- byte[] finish = predicate.getSlice_range().getFinish();
- if ((finish != null) && (finish.length > 0))
- return false;
-
- return true;
- }
-
- public void initialize(InputSplit split, TaskAttemptContext context) throws IOException
- {
- this.split = (ColumnFamilySplit) split;
- Configuration conf = HadoopCompat.getConfiguration(context);
- KeyRange jobRange = ConfigHelper.getInputKeyRange(conf);
- filter = jobRange == null ? null : jobRange.row_filter;
- predicate = ConfigHelper.getInputSlicePredicate(conf);
- boolean widerows = ConfigHelper.getInputIsWide(conf);
- isEmptyPredicate = isEmptyPredicate(predicate);
- totalRowCount = (this.split.getLength() < Long.MAX_VALUE)
- ? (int) this.split.getLength()
- : ConfigHelper.getInputSplitSize(conf);
- batchSize = ConfigHelper.getRangeBatchSize(conf);
- cfName = ConfigHelper.getInputColumnFamily(conf);
- consistencyLevel = ConsistencyLevel.valueOf(ConfigHelper.getReadConsistencyLevel(conf));
- keyspace = ConfigHelper.getInputKeyspace(conf);
-
- if (batchSize < 2)
- throw new IllegalArgumentException("Minimum batchSize is 2. Suggested batchSize is 100 or more");
-
- try
- {
- if (client != null)
- return;
-
- // create connection using thrift
- String location = getLocation();
-
- int port = ConfigHelper.getInputRpcPort(conf);
- client = ColumnFamilyInputFormat.createAuthenticatedClient(location, port, conf);
-
- }
- catch (Exception e)
- {
- throw new RuntimeException(e);
- }
-
- iter = widerows ? new WideRowIterator() : new StaticRowIterator();
- logger.debug("created {}", iter);
- }
-
- public boolean nextKeyValue() throws IOException
- {
- if (!iter.hasNext())
- {
- logger.debug("Finished scanning {} rows (estimate was: {})", iter.rowsRead(), totalRowCount);
- return false;
- }
-
- currentRow = iter.next();
- return true;
- }
-
- // we don't use endpointsnitch since we are trying to support hadoop nodes that are
- // not necessarily on Cassandra machines, too. This should be adequate for single-DC clusters, at least.
- private String getLocation()
- {
- Collection<InetAddress> localAddresses = FBUtilities.getAllLocalAddresses();
-
- for (InetAddress address : localAddresses)
- {
- for (String location : split.getLocations())
- {
- InetAddress locationAddress = null;
- try
- {
- locationAddress = InetAddress.getByName(location);
- }
- catch (UnknownHostException e)
- {
- throw new AssertionError(e);
- }
- if (address.equals(locationAddress))
- {
- return location;
- }
- }
- }
- return split.getLocations()[0];
- }
-
- private abstract class RowIterator extends AbstractIterator<Pair<ByteBuffer, SortedMap<ByteBuffer, Column>>>
- {
- protected List<KeySlice> rows;
- protected int totalRead = 0;
- protected final boolean isSuper;
- protected final AbstractType<?> comparator;
- protected final AbstractType<?> subComparator;
- protected final IPartitioner partitioner;
-
- private RowIterator()
- {
- CfDef cfDef = new CfDef();
- try
- {
- partitioner = FBUtilities.newPartitioner(client.describe_partitioner());
- // get CF meta data
- String query = String.format("SELECT comparator, subcomparator, type " +
- "FROM %s.%s " +
- "WHERE keyspace_name = '%s' AND columnfamily_name = '%s'",
- SystemKeyspace.NAME,
- LegacySchemaTables.COLUMNFAMILIES,
- keyspace,
- cfName);
-
- CqlResult result = client.execute_cql3_query(ByteBufferUtil.bytes(query), Compression.NONE, ConsistencyLevel.ONE);
-
- Iterator<CqlRow> iteraRow = result.rows.iterator();
-
- if (iteraRow.hasNext())
- {
- CqlRow cqlRow = iteraRow.next();
- cfDef.comparator_type = ByteBufferUtil.string(cqlRow.columns.get(0).value);
- ByteBuffer subComparator = cqlRow.columns.get(1).value;
- if (subComparator != null)
- cfDef.subcomparator_type = ByteBufferUtil.string(subComparator);
-
- ByteBuffer type = cqlRow.columns.get(2).value;
- if (type != null)
- cfDef.column_type = ByteBufferUtil.string(type);
- }
-
- comparator = TypeParser.parse(cfDef.comparator_type);
- subComparator = cfDef.subcomparator_type == null ? null : TypeParser.parse(cfDef.subcomparator_type);
- }
- catch (ConfigurationException e)
- {
- throw new RuntimeException("unable to load sub/comparator", e);
- }
- catch (TException e)
- {
- throw new RuntimeException("error communicating via Thrift", e);
- }
- catch (Exception e)
- {
- throw new RuntimeException("unable to load keyspace " + keyspace, e);
- }
- isSuper = "Super".equalsIgnoreCase(cfDef.column_type);
- }
-
- /**
- * @return total number of rows read by this record reader
- */
- public int rowsRead()
- {
- return totalRead;
- }
-
- protected List<Pair<ByteBuffer, Column>> unthriftify(ColumnOrSuperColumn cosc)
- {
- if (cosc.counter_column != null)
- return Collections.singletonList(unthriftifyCounter(cosc.counter_column));
- if (cosc.counter_super_column != null)
- return unthriftifySuperCounter(cosc.counter_super_column);
- if (cosc.super_column != null)
- return unthriftifySuper(cosc.super_column);
- assert cosc.column != null;
- return Collections.singletonList(unthriftifySimple(cosc.column));
- }
-
- private List<Pair<ByteBuffer, Column>> unthriftifySuper(SuperColumn super_column)
- {
- List<Pair<ByteBuffer, Column>> columns = new ArrayList<>(super_column.columns.size());
- for (org.apache.cassandra.thrift.Column column : super_column.columns)
- {
- Pair<ByteBuffer, Column> c = unthriftifySimple(column);
- columns.add(Pair.create(CompositeType.build(super_column.name, c.left), c.right));
- }
- return columns;
- }
-
- protected Pair<ByteBuffer, Column> unthriftifySimple(org.apache.cassandra.thrift.Column column)
- {
- return Pair.create(column.name, Column.fromRegularColumn(column));
- }
-
- private Pair<ByteBuffer, Column> unthriftifyCounter(CounterColumn column)
- {
- return Pair.create(column.name, Column.fromCounterColumn(column));
- }
-
- private List<Pair<ByteBuffer, Column>> unthriftifySuperCounter(CounterSuperColumn super_column)
- {
- List<Pair<ByteBuffer, Column>> columns = new ArrayList<>(super_column.columns.size());
- for (CounterColumn column : super_column.columns)
- {
- Pair<ByteBuffer, Column> c = unthriftifyCounter(column);
- columns.add(Pair.create(CompositeType.build(super_column.name, c.left), c.right));
- }
- return columns;
- }
- }
-
- private class StaticRowIterator extends RowIterator
- {
- protected int i = 0;
-
- private void maybeInit()
- {
- // check if we need another batch
- if (rows != null && i < rows.size())
- return;
-
- String startToken;
- if (totalRead == 0)
- {
- // first request
- startToken = split.getStartToken();
- }
- else
- {
- startToken = partitioner.getTokenFactory().toString(partitioner.getToken(Iterables.getLast(rows).key));
- if (startToken.equals(split.getEndToken()))
- {
- // reached end of the split
- rows = null;
- return;
- }
- }
-
- KeyRange keyRange = new KeyRange(batchSize)
- .setStart_token(startToken)
- .setEnd_token(split.getEndToken())
- .setRow_filter(filter);
- try
- {
- rows = client.get_range_slices(new ColumnParent(cfName), predicate, keyRange, consistencyLevel);
-
- // nothing new? reached the end
- if (rows.isEmpty())
- {
- rows = null;
- return;
- }
-
- // remove ghosts when fetching all columns
- if (isEmptyPredicate)
- {
- Iterator<KeySlice> it = rows.iterator();
- KeySlice ks;
- do
- {
- ks = it.next();
- if (ks.getColumnsSize() == 0)
- {
- it.remove();
- }
- } while (it.hasNext());
-
- // all ghosts, spooky
- if (rows.isEmpty())
- {
- // maybeInit assumes it can get the start-with key from the rows collection, so add back the last
- rows.add(ks);
- maybeInit();
- return;
- }
- }
-
- // reset to iterate through this new batch
- i = 0;
- }
- catch (Exception e)
- {
- throw new RuntimeException(e);
- }
- }
-
- protected Pair<ByteBuffer, SortedMap<ByteBuffer, Column>> computeNext()
- {
- maybeInit();
- if (rows == null)
- return endOfData();
-
- totalRead++;
- KeySlice ks = rows.get(i++);
- AbstractType<?> comp = isSuper ? CompositeType.getInstance(comparator, subComparator) : comparator;
- SortedMap<ByteBuffer, Column> map = new TreeMap<>(comp);
- for (ColumnOrSuperColumn cosc : ks.columns)
- {
- List<Pair<ByteBuffer, Column>> columns = unthriftify(cosc);
- for (Pair<ByteBuffer, Column> column : columns)
- map.put(column.left, column.right);
- }
- return Pair.create(ks.key, map);
- }
- }
-
- private class WideRowIterator extends RowIterator
- {
- private PeekingIterator<Pair<ByteBuffer, SortedMap<ByteBuffer, Column>>> wideColumns;
- private ByteBuffer lastColumn = ByteBufferUtil.EMPTY_BYTE_BUFFER;
- private ByteBuffer lastCountedKey = ByteBufferUtil.EMPTY_BYTE_BUFFER;
-
- private void maybeInit()
- {
- if (wideColumns != null && wideColumns.hasNext())
- return;
-
- KeyRange keyRange;
- if (totalRead == 0)
- {
- String startToken = split.getStartToken();
- keyRange = new KeyRange(batchSize)
- .setStart_token(startToken)
- .setEnd_token(split.getEndToken())
- .setRow_filter(filter);
- }
- else
- {
- KeySlice lastRow = Iterables.getLast(rows);
- logger.debug("Starting with last-seen row {}", lastRow.key);
- keyRange = new KeyRange(batchSize)
- .setStart_key(lastRow.key)
- .setEnd_token(split.getEndToken())
- .setRow_filter(filter);
- }
-
- try
- {
- rows = client.get_paged_slice(cfName, keyRange, lastColumn, consistencyLevel);
- int n = 0;
- for (KeySlice row : rows)
- n += row.columns.size();
- logger.debug("read {} columns in {} rows for {} starting with {}",
- new Object[]{ n, rows.size(), keyRange, lastColumn });
-
- wideColumns = Iterators.peekingIterator(new WideColumnIterator(rows));
- if (wideColumns.hasNext() && wideColumns.peek().right.keySet().iterator().next().equals(lastColumn))
- wideColumns.next();
- if (!wideColumns.hasNext())
- rows = null;
- }
- catch (Exception e)
- {
- throw new RuntimeException(e);
- }
- }
-
- protected Pair<ByteBuffer, SortedMap<ByteBuffer, Column>> computeNext()
- {
- maybeInit();
- if (rows == null)
- return endOfData();
-
- Pair<ByteBuffer, SortedMap<ByteBuffer, Column>> next = wideColumns.next();
- lastColumn = next.right.keySet().iterator().next().duplicate();
-
- maybeIncreaseRowCounter(next);
- return next;
- }
-
-
- /**
- * Increases the row counter only if we really moved to the next row.
- * @param next just fetched row slice
- */
- private void maybeIncreaseRowCounter(Pair<ByteBuffer, SortedMap<ByteBuffer, Column>> next)
- {
- ByteBuffer currentKey = next.left;
- if (!currentKey.equals(lastCountedKey))
- {
- totalRead++;
- lastCountedKey = currentKey;
- }
- }
-
- private class WideColumnIterator extends AbstractIterator<Pair<ByteBuffer, SortedMap<ByteBuffer, Column>>>
- {
- private final Iterator<KeySlice> rows;
- private Iterator<ColumnOrSuperColumn> columns;
- public KeySlice currentRow;
-
- public WideColumnIterator(List<KeySlice> rows)
- {
- this.rows = rows.iterator();
- if (this.rows.hasNext())
- nextRow();
- else
- columns = Iterators.emptyIterator();
- }
-
- private void nextRow()
- {
- currentRow = rows.next();
- columns = currentRow.columns.iterator();
- }
-
- protected Pair<ByteBuffer, SortedMap<ByteBuffer, Column>> computeNext()
- {
- AbstractType<?> comp = isSuper ? CompositeType.getInstance(comparator, subComparator) : comparator;
- while (true)
- {
- if (columns.hasNext())
- {
- ColumnOrSuperColumn cosc = columns.next();
- SortedMap<ByteBuffer, Column> map;
- List<Pair<ByteBuffer, Column>> columns = unthriftify(cosc);
- if (columns.size() == 1)
- {
- map = ImmutableSortedMap.of(columns.get(0).left, columns.get(0).right);
- }
- else
- {
- assert isSuper;
- map = new TreeMap<>(comp);
- for (Pair<ByteBuffer, Column> column : columns)
- map.put(column.left, column.right);
- }
- return Pair.create(currentRow.key, map);
- }
-
- if (!rows.hasNext())
- return endOfData();
-
- nextRow();
- }
- }
- }
- }
-
- // Because the old Hadoop API wants us to write to the key and value
- // and the new asks for them, we need to copy the output of the new API
- // to the old. Thus, expect a small performance hit.
- // And obviously this wouldn't work for wide rows. But since ColumnFamilyInputFormat
- // and ColumnFamilyRecordReader don't support them, it should be fine for now.
- public boolean next(ByteBuffer key, SortedMap<ByteBuffer, Column> value) throws IOException
- {
- if (this.nextKeyValue())
- {
- key.clear();
- key.put(this.getCurrentKey().duplicate());
- key.flip();
-
- value.clear();
- value.putAll(this.getCurrentValue());
-
- return true;
- }
- return false;
- }
-
- public ByteBuffer createKey()
- {
- return ByteBuffer.wrap(new byte[this.keyBufferSize]);
- }
-
- public SortedMap<ByteBuffer, Column> createValue()
- {
- return new TreeMap<>();
- }
-
- public long getPos() throws IOException
- {
- return iter.rowsRead();
- }
-
- public static final class Column
- {
- public final ByteBuffer name;
- public final ByteBuffer value;
- public final long timestamp;
-
- private Column(ByteBuffer name, ByteBuffer value, long timestamp)
- {
- this.name = name;
- this.value = value;
- this.timestamp = timestamp;
- }
-
- static Column fromRegularColumn(org.apache.cassandra.thrift.Column input)
- {
- return new Column(input.name, input.value, input.timestamp);
- }
-
- static Column fromCounterColumn(org.apache.cassandra.thrift.CounterColumn input)
- {
- return new Column(input.name, ByteBufferUtil.bytes(input.value), 0);
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/446e2537/src/java/org/apache/cassandra/hadoop/ColumnFamilyRecordWriter.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/hadoop/ColumnFamilyRecordWriter.java b/src/java/org/apache/cassandra/hadoop/ColumnFamilyRecordWriter.java
deleted file mode 100644
index 9547d0e..0000000
--- a/src/java/org/apache/cassandra/hadoop/ColumnFamilyRecordWriter.java
+++ /dev/null
@@ -1,341 +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.cassandra.hadoop;
-
-
-import java.io.IOException;
-import java.net.InetAddress;
-import java.nio.ByteBuffer;
-import java.util.*;
-import java.util.concurrent.*;
-
-import org.apache.cassandra.client.*;
-import org.apache.cassandra.dht.Range;
-import org.apache.cassandra.dht.Token;
-import org.apache.cassandra.thrift.*;
-import org.apache.cassandra.thrift.ConsistencyLevel;
-import org.apache.cassandra.utils.*;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.mapreduce.*;
-import org.apache.thrift.TException;
-import org.apache.hadoop.util.Progressable;
-import org.apache.thrift.transport.*;
-
-
-/**
- * The <code>ColumnFamilyRecordWriter</code> maps the output <key, value>
- * pairs to a Cassandra column family. In particular, it applies all mutations
- * in the value, which it associates with the key, and in turn the responsible
- * endpoint.
- *
- * <p>
- * Furthermore, this writer groups the mutations by the endpoint responsible for
- * the rows being affected. This allows the mutations to be executed in parallel,
- * directly to a responsible endpoint.
- * </p>
- *
- * @see ColumnFamilyOutputFormat
- */
-@Deprecated
-final class ColumnFamilyRecordWriter extends RecordWriter<ByteBuffer, List<Mutation>> implements
- org.apache.hadoop.mapred.RecordWriter<ByteBuffer, List<Mutation>>
-{
- // The configuration this writer is associated with.
- protected final Configuration conf;
-
- // The number of mutations to buffer per endpoint
- protected final int queueSize;
-
- protected final long batchThreshold;
-
- protected final ConsistencyLevel consistencyLevel;
- protected Progressable progressable;
- protected TaskAttemptContext context;
- // handles for clients for each range running in the threadpool
- private final Map<Range, RangeClient> clients;
-
- // The ring cache that describes the token ranges each node in the ring is
- // responsible for. This is what allows us to group the mutations by
- // the endpoints they should be targeted at. The targeted endpoint
- // essentially
- // acts as the primary replica for the rows being affected by the mutations.
- private final RingCache ringCache;
-
- /**
- * Upon construction, obtain the map that this writer will use to collect
- * mutations, and the ring cache for the given keyspace.
- *
- * @param context the task attempt context
- * @throws IOException
- */
- ColumnFamilyRecordWriter(TaskAttemptContext context)
- {
- this(HadoopCompat.getConfiguration(context));
- this.context = context;
-
- }
- ColumnFamilyRecordWriter(Configuration conf, Progressable progressable)
- {
- this(conf);
- this.progressable = progressable;
- }
-
- ColumnFamilyRecordWriter(Configuration conf)
- {
- this.conf = conf;
- this.queueSize = conf.getInt(ColumnFamilyOutputFormat.QUEUE_SIZE, 32 * FBUtilities.getAvailableProcessors());
- batchThreshold = conf.getLong(ColumnFamilyOutputFormat.BATCH_THRESHOLD, 32);
- consistencyLevel = ConsistencyLevel.valueOf(ConfigHelper.getWriteConsistencyLevel(conf));
- this.ringCache = new RingCache(conf);
- this.clients = new HashMap<Range, RangeClient>();
- }
-
- /**
- * Close this <code>RecordWriter</code> to future operations, but not before
- * flushing out the batched mutations.
- *
- * @param context the context of the task
- * @throws IOException
- */
- public void close(TaskAttemptContext context) throws IOException
- {
- close();
- }
-
- /** Fills the deprecated RecordWriter interface for streaming. */
- @Deprecated
- public void close(org.apache.hadoop.mapred.Reporter reporter) throws IOException
- {
- close();
- }
-
- public void close() throws IOException
- {
- // close all the clients before throwing anything
- IOException clientException = null;
- for (RangeClient client : clients.values())
- {
- try
- {
- client.close();
- }
- catch (IOException e)
- {
- clientException = e;
- }
- }
- if (clientException != null)
- throw clientException;
- }
-
- /**
- * If the key is to be associated with a valid value, a mutation is created
- * for it with the given column family and columns. In the event the value
- * in the column is missing (i.e., null), then it is marked for
- * {@link Deletion}. Similarly, if the entire value for a key is missing
- * (i.e., null), then the entire key is marked for {@link Deletion}.
- * </p>
- *
- * @param keybuff
- * the key to write.
- * @param value
- * the value to write.
- * @throws IOException
- */
- @Override
- public void write(ByteBuffer keybuff, List<Mutation> value) throws IOException
- {
- Range<Token> range = ringCache.getRange(keybuff);
-
- // get the client for the given range, or create a new one
- RangeClient client = clients.get(range);
- if (client == null)
- {
- // haven't seen keys for this range: create new client
- client = new RangeClient(ringCache.getEndpoint(range));
- client.start();
- clients.put(range, client);
- }
-
- for (Mutation amut : value)
- client.put(Pair.create(keybuff, amut));
- if (progressable != null)
- progressable.progress();
- if (context != null)
- HadoopCompat.progress(context);
- }
-
- /**
- * A client that runs in a threadpool and connects to the list of endpoints for a particular
- * range. Mutations for keys in that range are sent to this client via a queue.
- */
- public class RangeClient extends Thread
- {
- // The list of endpoints for this range
- protected final List<InetAddress> endpoints;
- // A bounded queue of incoming mutations for this range
- protected final BlockingQueue<Pair<ByteBuffer, Mutation>> queue = new ArrayBlockingQueue<>(queueSize);
-
- protected volatile boolean run = true;
- // we want the caller to know if something went wrong, so we record any unrecoverable exception while writing
- // so we can throw it on the caller's stack when he calls put() again, or if there are no more put calls,
- // when the client is closed.
- protected volatile IOException lastException;
-
- protected Cassandra.Client client;
- public final String columnFamily = ConfigHelper.getOutputColumnFamily(conf);
-
- /**
- * Constructs an {@link RangeClient} for the given endpoints.
- * @param endpoints the possible endpoints to execute the mutations on
- */
- public RangeClient(List<InetAddress> endpoints)
- {
- super("client-" + endpoints);
- this.endpoints = endpoints;
- }
-
- /**
- * enqueues the given value to Cassandra
- */
- public void put(Pair<ByteBuffer, Mutation> value) throws IOException
- {
- while (true)
- {
- if (lastException != null)
- throw lastException;
- try
- {
- if (queue.offer(value, 100, TimeUnit.MILLISECONDS))
- break;
- }
- catch (InterruptedException e)
- {
- throw new AssertionError(e);
- }
- }
- }
-
- public void close() throws IOException
- {
- // stop the run loop. this will result in closeInternal being called by the time join() finishes.
- run = false;
- interrupt();
- try
- {
- this.join();
- }
- catch (InterruptedException e)
- {
- throw new AssertionError(e);
- }
-
- if (lastException != null)
- throw lastException;
- }
-
- @SuppressWarnings("resource")
- protected void closeInternal()
- {
- if (client != null)
- {
- TTransport transport = client.getOutputProtocol().getTransport();
- if (transport.isOpen())
- transport.close();
- }
- }
-
- /**
- * Loops collecting mutations from the queue and sending to Cassandra
- */
- public void run()
- {
- outer:
- while (run || !queue.isEmpty())
- {
- Pair<ByteBuffer, Mutation> mutation;
- try
- {
- mutation = queue.take();
- }
- catch (InterruptedException e)
- {
- // re-check loop condition after interrupt
- continue;
- }
-
- Map<ByteBuffer, Map<String, List<Mutation>>> batch = new HashMap<ByteBuffer, Map<String, List<Mutation>>>();
- while (mutation != null)
- {
- Map<String, List<Mutation>> subBatch = batch.get(mutation.left);
- if (subBatch == null)
- {
- subBatch = Collections.singletonMap(columnFamily, (List<Mutation>) new ArrayList<Mutation>());
- batch.put(mutation.left, subBatch);
- }
-
- subBatch.get(columnFamily).add(mutation.right);
- if (batch.size() >= batchThreshold)
- break;
-
- mutation = queue.poll();
- }
-
- Iterator<InetAddress> iter = endpoints.iterator();
- while (true)
- {
- // send the mutation to the last-used endpoint. first time through, this will NPE harmlessly.
- try
- {
- client.batch_mutate(batch, consistencyLevel);
- break;
- }
- catch (Exception e)
- {
- closeInternal();
- if (!iter.hasNext())
- {
- lastException = new IOException(e);
- break outer;
- }
- }
-
- // attempt to connect to a different endpoint
- try
- {
- InetAddress address = iter.next();
- String host = address.getHostName();
- int port = ConfigHelper.getOutputRpcPort(conf);
- client = ColumnFamilyOutputFormat.createAuthenticatedClient(host, port, conf);
- }
- catch (Exception e)
- {
- closeInternal();
- // TException means something unexpected went wrong to that endpoint, so
- // we should try again to another. Other exceptions (auth or invalid request) are fatal.
- if ((!(e instanceof TException)) || !iter.hasNext())
- {
- lastException = new IOException(e);
- break outer;
- }
- }
- }
- }
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/446e2537/src/java/org/apache/cassandra/hadoop/cql3/CqlBulkRecordWriter.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/hadoop/cql3/CqlBulkRecordWriter.java b/src/java/org/apache/cassandra/hadoop/cql3/CqlBulkRecordWriter.java
index e77c4c8..3e69c2d 100644
--- a/src/java/org/apache/cassandra/hadoop/cql3/CqlBulkRecordWriter.java
+++ b/src/java/org/apache/cassandra/hadoop/cql3/CqlBulkRecordWriter.java
@@ -33,7 +33,6 @@ import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.config.Config;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.exceptions.InvalidRequestException;
-import org.apache.cassandra.hadoop.BulkRecordWriter;
import org.apache.cassandra.hadoop.ConfigHelper;
import org.apache.cassandra.hadoop.HadoopCompat;
import org.apache.cassandra.io.sstable.CQLSSTableWriter;
@@ -41,6 +40,7 @@ import org.apache.cassandra.io.sstable.SSTableLoader;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.streaming.StreamState;
import org.apache.cassandra.utils.NativeSSTableLoaderClient;
+import org.apache.cassandra.utils.OutputHandler;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
@@ -152,7 +152,7 @@ public class CqlBulkRecordWriter extends RecordWriter<Object, List<ByteBuffer>>
ExternalClient externalClient = new ExternalClient(conf);
externalClient.setTableMetadata(CFMetaData.compile(schema, keyspace));
- loader = new SSTableLoader(outputDir, externalClient, new BulkRecordWriter.NullOutputHandler())
+ loader = new SSTableLoader(outputDir, externalClient, new NullOutputHandler())
{
@Override
public void onSuccess(StreamState finalState)
@@ -287,4 +287,12 @@ public class CqlBulkRecordWriter extends RecordWriter<Object, List<ByteBuffer>>
return addresses;
}
}
+
+ public static class NullOutputHandler implements OutputHandler
+ {
+ public void output(String msg) {}
+ public void debug(String msg) {}
+ public void warn(String msg) {}
+ public void warn(String msg, Throwable th) {}
+ }
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/446e2537/src/java/org/apache/cassandra/hadoop/cql3/CqlInputFormat.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/hadoop/cql3/CqlInputFormat.java b/src/java/org/apache/cassandra/hadoop/cql3/CqlInputFormat.java
index 09bd80c..70429a8 100644
--- a/src/java/org/apache/cassandra/hadoop/cql3/CqlInputFormat.java
+++ b/src/java/org/apache/cassandra/hadoop/cql3/CqlInputFormat.java
@@ -18,12 +18,44 @@
package org.apache.cassandra.hadoop.cql3;
import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+import java.util.Set;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
-import org.apache.cassandra.hadoop.AbstractColumnFamilyInputFormat;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.datastax.driver.core.Host;
+import com.datastax.driver.core.Metadata;
+import com.datastax.driver.core.ResultSet;
+import com.datastax.driver.core.Session;
+import com.datastax.driver.core.TokenRange;
+import org.apache.cassandra.db.SystemKeyspace;
+import org.apache.cassandra.dht.ByteOrderedPartitioner;
+import org.apache.cassandra.dht.IPartitioner;
+import org.apache.cassandra.dht.OrderPreservingPartitioner;
+import org.apache.cassandra.dht.Range;
+import org.apache.cassandra.dht.Token;
+import org.apache.cassandra.hadoop.ColumnFamilySplit;
+import org.apache.cassandra.hadoop.ConfigHelper;
+import org.apache.cassandra.hadoop.HadoopCompat;
+import org.apache.cassandra.thrift.KeyRange;
+import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.mapred.InputSplit;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.RecordReader;
import org.apache.hadoop.mapred.Reporter;
+import org.apache.hadoop.mapreduce.JobContext;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.TaskAttemptID;
@@ -48,8 +80,15 @@ import com.datastax.driver.core.Row;
*
* other native protocol connection parameters in CqlConfigHelper
*/
-public class CqlInputFormat extends AbstractColumnFamilyInputFormat<Long, Row>
+public class CqlInputFormat extends org.apache.hadoop.mapreduce.InputFormat<Long, Row> implements org.apache.hadoop.mapred.InputFormat<Long, Row>
{
+ public static final String MAPRED_TASK_ID = "mapred.task.id";
+ private static final Logger logger = LoggerFactory.getLogger(CqlInputFormat.class);
+ private String keyspace;
+ private String cfName;
+ private IPartitioner partitioner;
+ private Session session;
+
public RecordReader<Long, Row> getRecordReader(InputSplit split, JobConf jobConf, final Reporter reporter)
throws IOException
{
@@ -75,4 +114,238 @@ public class CqlInputFormat extends AbstractColumnFamilyInputFormat<Long, Row>
return new CqlRecordReader();
}
+ protected void validateConfiguration(Configuration conf)
+ {
+ if (ConfigHelper.getInputKeyspace(conf) == null || ConfigHelper.getInputColumnFamily(conf) == null)
+ {
+ throw new UnsupportedOperationException("you must set the keyspace and table with setInputColumnFamily()");
+ }
+ if (ConfigHelper.getInputInitialAddress(conf) == null)
+ throw new UnsupportedOperationException("You must set the initial output address to a Cassandra node with setInputInitialAddress");
+ if (ConfigHelper.getInputPartitioner(conf) == null)
+ throw new UnsupportedOperationException("You must set the Cassandra partitioner class with setInputPartitioner");
+ }
+
+ public List<org.apache.hadoop.mapreduce.InputSplit> getSplits(JobContext context) throws IOException
+ {
+ Configuration conf = HadoopCompat.getConfiguration(context);
+
+ validateConfiguration(conf);
+
+ keyspace = ConfigHelper.getInputKeyspace(conf);
+ cfName = ConfigHelper.getInputColumnFamily(conf);
+ partitioner = ConfigHelper.getInputPartitioner(conf);
+ logger.debug("partitioner is {}", partitioner);
+
+ // canonical ranges and nodes holding replicas
+ Map<TokenRange, Set<Host>> masterRangeNodes = getRangeMap(conf, keyspace);
+
+ // canonical ranges, split into pieces, fetching the splits in parallel
+ ExecutorService executor = new ThreadPoolExecutor(0, 128, 60L, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>());
+ List<org.apache.hadoop.mapreduce.InputSplit> splits = new ArrayList<>();
+
+ try
+ {
+ List<Future<List<org.apache.hadoop.mapreduce.InputSplit>>> splitfutures = new ArrayList<>();
+ KeyRange jobKeyRange = ConfigHelper.getInputKeyRange(conf);
+ Range<Token> jobRange = null;
+ if (jobKeyRange != null)
+ {
+ if (jobKeyRange.start_key != null)
+ {
+ if (!partitioner.preservesOrder())
+ throw new UnsupportedOperationException("KeyRange based on keys can only be used with a order preserving partitioner");
+ if (jobKeyRange.start_token != null)
+ throw new IllegalArgumentException("only start_key supported");
+ if (jobKeyRange.end_token != null)
+ throw new IllegalArgumentException("only start_key supported");
+ jobRange = new Range<>(partitioner.getToken(jobKeyRange.start_key),
+ partitioner.getToken(jobKeyRange.end_key));
+ }
+ else if (jobKeyRange.start_token != null)
+ {
+ jobRange = new Range<>(partitioner.getTokenFactory().fromString(jobKeyRange.start_token),
+ partitioner.getTokenFactory().fromString(jobKeyRange.end_token));
+ }
+ else
+ {
+ logger.warn("ignoring jobKeyRange specified without start_key or start_token");
+ }
+ }
+
+ session = CqlConfigHelper.getInputCluster(ConfigHelper.getInputInitialAddress(conf).split(","), conf).connect();
+ Metadata metadata = session.getCluster().getMetadata();
+
+ for (TokenRange range : masterRangeNodes.keySet())
+ {
+ if (jobRange == null)
+ {
+ // for each tokenRange, pick a live owner and ask it to compute bite-sized splits
+ splitfutures.add(executor.submit(new SplitCallable(range, masterRangeNodes.get(range), conf)));
+ }
+ else
+ {
+ TokenRange jobTokenRange = rangeToTokenRange(metadata, jobRange);
+ if (range.intersects(jobTokenRange))
+ {
+ for (TokenRange intersection: range.intersectWith(jobTokenRange))
+ {
+ // for each tokenRange, pick a live owner and ask it to compute bite-sized splits
+ splitfutures.add(executor.submit(new SplitCallable(intersection, masterRangeNodes.get(range), conf)));
+ }
+ }
+ }
+ }
+
+ // wait until we have all the results back
+ for (Future<List<org.apache.hadoop.mapreduce.InputSplit>> futureInputSplits : splitfutures)
+ {
+ try
+ {
+ splits.addAll(futureInputSplits.get());
+ }
+ catch (Exception e)
+ {
+ throw new IOException("Could not get input splits", e);
+ }
+ }
+ }
+ finally
+ {
+ executor.shutdownNow();
+ }
+
+ assert splits.size() > 0;
+ Collections.shuffle(splits, new Random(System.nanoTime()));
+ return splits;
+ }
+
+ private TokenRange rangeToTokenRange(Metadata metadata, Range<Token> range)
+ {
+ return metadata.newTokenRange(metadata.newToken(partitioner.getTokenFactory().toString(range.left)),
+ metadata.newToken(partitioner.getTokenFactory().toString(range.right)));
+ }
+
+ private Map<TokenRange, Long> getSubSplits(String keyspace, String cfName, TokenRange range, Configuration conf) throws IOException
+ {
+ int splitSize = ConfigHelper.getInputSplitSize(conf);
+ try
+ {
+ return describeSplits(keyspace, cfName, range, splitSize);
+ }
+ catch (Exception e)
+ {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private Map<TokenRange, Set<Host>> getRangeMap(Configuration conf, String keyspace)
+ {
+ try (Session session = CqlConfigHelper.getInputCluster(ConfigHelper.getInputInitialAddress(conf).split(","), conf).connect())
+ {
+ Map<TokenRange, Set<Host>> map = new HashMap<>();
+ Metadata metadata = session.getCluster().getMetadata();
+ for (TokenRange tokenRange : metadata.getTokenRanges())
+ map.put(tokenRange, metadata.getReplicas('"' + keyspace + '"', tokenRange));
+ return map;
+ }
+ }
+
+ private Map<TokenRange, Long> describeSplits(String keyspace, String table, TokenRange tokenRange, int splitSize)
+ {
+ String query = String.format("SELECT mean_partition_size, partitions_count " +
+ "FROM %s.%s " +
+ "WHERE keyspace_name = ? AND table_name = ? AND range_start = ? AND range_end = ?",
+ SystemKeyspace.NAME,
+ SystemKeyspace.SIZE_ESTIMATES);
+
+ ResultSet resultSet = session.execute(query, keyspace, table, tokenRange.getStart().toString(), tokenRange.getEnd().toString());
+
+ Row row = resultSet.one();
+ // If we have no data on this split, return the full split i.e., do not sub-split
+ // Assume smallest granularity of partition count available from CASSANDRA-7688
+ if (row == null)
+ {
+ Map<TokenRange, Long> wrappedTokenRange = new HashMap<>();
+ wrappedTokenRange.put(tokenRange, (long) 128);
+ return wrappedTokenRange;
+ }
+
+ long meanPartitionSize = row.getLong("mean_partition_size");
+ long partitionCount = row.getLong("partitions_count");
+
+ int splitCount = (int)((meanPartitionSize * partitionCount) / splitSize);
+ List<TokenRange> splitRanges = tokenRange.splitEvenly(splitCount);
+ Map<TokenRange, Long> rangesWithLength = new HashMap<>();
+ for (TokenRange range : splitRanges)
+ rangesWithLength.put(range, partitionCount/splitCount);
+
+ return rangesWithLength;
+ }
+
+ // Old Hadoop API
+ public InputSplit[] getSplits(JobConf jobConf, int numSplits) throws IOException
+ {
+ TaskAttemptContext tac = HadoopCompat.newTaskAttemptContext(jobConf, new TaskAttemptID());
+ List<org.apache.hadoop.mapreduce.InputSplit> newInputSplits = this.getSplits(tac);
+ InputSplit[] oldInputSplits = new InputSplit[newInputSplits.size()];
+ for (int i = 0; i < newInputSplits.size(); i++)
+ oldInputSplits[i] = (ColumnFamilySplit)newInputSplits.get(i);
+ return oldInputSplits;
+ }
+
+ /**
+ * Gets a token tokenRange and splits it up according to the suggested
+ * size into input splits that Hadoop can use.
+ */
+ class SplitCallable implements Callable<List<org.apache.hadoop.mapreduce.InputSplit>>
+ {
+
+ private final TokenRange tokenRange;
+ private final Set<Host> hosts;
+ private final Configuration conf;
+
+ public SplitCallable(TokenRange tr, Set<Host> hosts, Configuration conf)
+ {
+ this.tokenRange = tr;
+ this.hosts = hosts;
+ this.conf = conf;
+ }
+
+ public List<org.apache.hadoop.mapreduce.InputSplit> call() throws Exception
+ {
+ ArrayList<org.apache.hadoop.mapreduce.InputSplit> splits = new ArrayList<>();
+ Map<TokenRange, Long> subSplits;
+ subSplits = getSubSplits(keyspace, cfName, tokenRange, conf);
+ // turn the sub-ranges into InputSplits
+ String[] endpoints = new String[hosts.size()];
+
+ // hadoop needs hostname, not ip
+ int endpointIndex = 0;
+ for (Host endpoint : hosts)
+ endpoints[endpointIndex++] = endpoint.getAddress().getHostName();
+
+ boolean partitionerIsOpp = partitioner instanceof OrderPreservingPartitioner || partitioner instanceof ByteOrderedPartitioner;
+
+ for (TokenRange subSplit : subSplits.keySet())
+ {
+ List<TokenRange> ranges = subSplit.unwrap();
+ for (TokenRange subrange : ranges)
+ {
+ ColumnFamilySplit split =
+ new ColumnFamilySplit(
+ partitionerIsOpp ?
+ subrange.getStart().toString().substring(2) : subrange.getStart().toString(),
+ partitionerIsOpp ?
+ subrange.getEnd().toString().substring(2) : subrange.getStart().toString(),
+ subSplits.get(subSplit),
+ endpoints);
+
+ logger.debug("adding {}", split);
+ splits.add(split);
+ }
+ }
+ return splits;
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/446e2537/src/java/org/apache/cassandra/hadoop/cql3/CqlOutputFormat.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/hadoop/cql3/CqlOutputFormat.java b/src/java/org/apache/cassandra/hadoop/cql3/CqlOutputFormat.java
index 9a1cda6..cc0a6b1 100644
--- a/src/java/org/apache/cassandra/hadoop/cql3/CqlOutputFormat.java
+++ b/src/java/org/apache/cassandra/hadoop/cql3/CqlOutputFormat.java
@@ -55,6 +55,9 @@ import org.apache.hadoop.mapreduce.*;
public class CqlOutputFormat extends OutputFormat<Map<String, ByteBuffer>, List<ByteBuffer>>
implements org.apache.hadoop.mapred.OutputFormat<Map<String, ByteBuffer>, List<ByteBuffer>>
{
+ public static final String BATCH_THRESHOLD = "mapreduce.output.columnfamilyoutputformat.batch.threshold";
+ public static final String QUEUE_SIZE = "mapreduce.output.columnfamilyoutputformat.queue.size";
+
/**
* Check for validity of the output-specification for the job.
*
http://git-wip-us.apache.org/repos/asf/cassandra/blob/446e2537/src/java/org/apache/cassandra/hadoop/cql3/CqlRecordWriter.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/hadoop/cql3/CqlRecordWriter.java b/src/java/org/apache/cassandra/hadoop/cql3/CqlRecordWriter.java
index 78b0494..c9198c6 100644
--- a/src/java/org/apache/cassandra/hadoop/cql3/CqlRecordWriter.java
+++ b/src/java/org/apache/cassandra/hadoop/cql3/CqlRecordWriter.java
@@ -31,7 +31,6 @@ import com.datastax.driver.core.exceptions.*;
import org.apache.cassandra.db.marshal.CompositeType;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.dht.Token;
-import org.apache.cassandra.hadoop.ColumnFamilyOutputFormat;
import org.apache.cassandra.hadoop.ConfigHelper;
import org.apache.cassandra.hadoop.HadoopCompat;
import org.apache.cassandra.utils.FBUtilities;
@@ -109,8 +108,8 @@ class CqlRecordWriter extends RecordWriter<Map<String, ByteBuffer>, List<ByteBuf
CqlRecordWriter(Configuration conf)
{
this.conf = conf;
- this.queueSize = conf.getInt(ColumnFamilyOutputFormat.QUEUE_SIZE, 32 * FBUtilities.getAvailableProcessors());
- batchThreshold = conf.getLong(ColumnFamilyOutputFormat.BATCH_THRESHOLD, 32);
+ this.queueSize = conf.getInt(CqlOutputFormat.QUEUE_SIZE, 32 * FBUtilities.getAvailableProcessors());
+ batchThreshold = conf.getLong(CqlOutputFormat.BATCH_THRESHOLD, 32);
this.clients = new HashMap<>();
try
[4/4] cassandra git commit: Remove deprecated legacy Hadoop code
Posted by al...@apache.org.
Remove deprecated legacy Hadoop code
patch by Philip Thompson; reviewed by Aleksey Yeschenko for
CASSANDRA-9353
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/446e2537
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/446e2537
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/446e2537
Branch: refs/heads/trunk
Commit: 446e2537895c15b404a74107069a12f3fc404b15
Parents: d62cd1b
Author: Philip Thompson <pt...@gmail.com>
Authored: Mon Jun 8 22:41:28 2015 +0300
Committer: Aleksey Yeschenko <al...@apache.org>
Committed: Mon Jun 8 22:41:28 2015 +0300
----------------------------------------------------------------------
CHANGES.txt | 1 +
NEWS.txt | 14 +
examples/hadoop_word_count/README.txt | 50 -
examples/hadoop_word_count/bin/word_count | 61 -
.../hadoop_word_count/bin/word_count_counters | 59 -
examples/hadoop_word_count/bin/word_count_setup | 61 -
examples/hadoop_word_count/build.xml | 113 --
examples/hadoop_word_count/conf/logback.xml | 42 -
examples/hadoop_word_count/ivy.xml | 24 -
examples/hadoop_word_count/src/WordCount.java | 222 ---
.../src/WordCountCounters.java | 104 --
.../hadoop_word_count/src/WordCountSetup.java | 239 ---
.../hadoop/AbstractColumnFamilyInputFormat.java | 296 ----
.../cassandra/hadoop/BulkOutputFormat.java | 91 --
.../cassandra/hadoop/BulkRecordWriter.java | 296 ----
.../hadoop/ColumnFamilyInputFormat.java | 125 --
.../hadoop/ColumnFamilyOutputFormat.java | 183 ---
.../hadoop/ColumnFamilyRecordReader.java | 615 --------
.../hadoop/ColumnFamilyRecordWriter.java | 341 -----
.../hadoop/cql3/CqlBulkRecordWriter.java | 12 +-
.../cassandra/hadoop/cql3/CqlInputFormat.java | 277 +++-
.../cassandra/hadoop/cql3/CqlOutputFormat.java | 3 +
.../cassandra/hadoop/cql3/CqlRecordWriter.java | 5 +-
.../cassandra/hadoop/pig/CassandraStorage.java | 1397 ------------------
.../org/apache/cassandra/pig/PigTestBase.java | 27 +-
.../pig/ThriftColumnFamilyDataTypeTest.java | 70 +-
.../cassandra/pig/ThriftColumnFamilyTest.java | 320 ++--
27 files changed, 451 insertions(+), 4597 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/446e2537/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 560bd2b..061bd5f 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
3.0:
+ * Remove deprecated legacy Hadoop code (CASSANDRA-9353)
* Decommissioned nodes will not rejoin the cluster (CASSANDRA-8801)
* Change gossip stabilization to use endpoit size (CASSANDRA-9401)
* Change default garbage collector to G1 (CASSANDRA-7486)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/446e2537/NEWS.txt
----------------------------------------------------------------------
diff --git a/NEWS.txt b/NEWS.txt
index 9beb911..44637eb 100644
--- a/NEWS.txt
+++ b/NEWS.txt
@@ -13,6 +13,20 @@ restore snapshots created with the previous major version using the
'sstableloader' tool. You can upgrade the file format of your snapshots
using the provided 'sstableupgrade' tool.
+3.0
+===
+
+Upgrading
+---------
+ - Pig's CassandraStorage has been removed. Use CqlNativeStorage instead.
+ - Hadoop BulkOutputFormat and BulkRecordWriter have been removed; use
+ CqlBulkOutputFormat and CqlBulkRecordWriter instead.
+ - Hadoop ColumnFamilyInputFormat and ColumnFamilyOutputFormat have been removed;
+ use CqlInputFormat and CqlOutputFormat instead.
+ - Hadoop ColumnFamilyRecordReader and ColumnFamilyRecordWriter have been removed;
+ use CqlRecordReader and CqlRecordWriter instead.
+
+
2.2
===
http://git-wip-us.apache.org/repos/asf/cassandra/blob/446e2537/examples/hadoop_word_count/README.txt
----------------------------------------------------------------------
diff --git a/examples/hadoop_word_count/README.txt b/examples/hadoop_word_count/README.txt
deleted file mode 100644
index e336b89..0000000
--- a/examples/hadoop_word_count/README.txt
+++ /dev/null
@@ -1,50 +0,0 @@
-Introduction
-============
-
-WordCount hadoop example: Inserts a bunch of words across multiple rows,
-and counts them, with RandomPartitioner. The word_count_counters example sums
-the value of counter columns for a key.
-
-The scripts in bin/ assume you are running with cwd of examples/word_count.
-
-
-Running
-=======
-
-First build and start a Cassandra server with the default configuration*. Ensure that the Thrift
-interface is enabled, either by setting start_rpc:true in cassandra.yaml or by running
-`nodetool enablethrift` after startup.
-Once Cassandra has started and the Thrift interface is available, run
-
-contrib/word_count$ ant
-contrib/word_count$ bin/word_count_setup
-contrib/word_count$ bin/word_count
-contrib/word_count$ bin/word_count_counters
-
-In order to view the results in Cassandra, one can use bin/cqlsh and
-perform the following operations:
-$ bin/cqlsh localhost
-> use wordcount;
-> select * from output_words;
-
-The output of the word count can now be configured. In the bin/word_count
-file, you can specify the OUTPUT_REDUCER. The two options are 'filesystem'
-and 'cassandra'. The filesystem option outputs to the /tmp/word_count*
-directories. The cassandra option outputs to the 'output_words' column family
-in the 'wordcount' keyspace. 'cassandra' is the default.
-
-Read the code in src/ for more details.
-
-The word_count_counters example sums the counter columns for a row. The output
-is written to a text file in /tmp/word_count_counters.
-
-*It is recommended to turn off vnodes when running Cassandra with hadoop.
-This is done by setting "num_tokens: 1" in cassandra.yaml. If you want to
-point wordcount at a real cluster, modify the seed and listenaddress
-settings accordingly.
-
-
-Troubleshooting
-===============
-
-word_count uses conf/logback.xml to log to wc.out.
http://git-wip-us.apache.org/repos/asf/cassandra/blob/446e2537/examples/hadoop_word_count/bin/word_count
----------------------------------------------------------------------
diff --git a/examples/hadoop_word_count/bin/word_count b/examples/hadoop_word_count/bin/word_count
deleted file mode 100755
index 34534d7..0000000
--- a/examples/hadoop_word_count/bin/word_count
+++ /dev/null
@@ -1,61 +0,0 @@
-#!/bin/sh
-
-# 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.
-
-cwd=`dirname $0`
-
-# Cassandra class files.
-if [ ! -d $cwd/../../../build/classes/main ]; then
- echo "Unable to locate cassandra class files" >&2
- exit 1
-fi
-
-# word_count Jar.
-if [ ! -e $cwd/../build/word_count.jar ]; then
- echo "Unable to locate word_count jar" >&2
- exit 1
-fi
-
-CLASSPATH=$CLASSPATH:$cwd/../conf
-CLASSPATH=$CLASSPATH:$cwd/../build/word_count.jar
-CLASSPATH=$CLASSPATH:$cwd/../../../build/classes/main
-CLASSPATH=$CLASSPATH:$cwd/../../../build/classes/thrift
-for jar in $cwd/../build/lib/jars/*.jar; do
- CLASSPATH=$CLASSPATH:$jar
-done
-for jar in $cwd/../../../lib/*.jar; do
- CLASSPATH=$CLASSPATH:$jar
-done
-for jar in $cwd/../../../build/lib/jars/*.jar; do
- CLASSPATH=$CLASSPATH:$jar
-done
-
-if [ -x $JAVA_HOME/bin/java ]; then
- JAVA=$JAVA_HOME/bin/java
-else
- JAVA=`which java`
-fi
-
-if [ "x$JAVA" = "x" ]; then
- echo "Java executable not found (hint: set JAVA_HOME)" >&2
- exit 1
-fi
-
-OUTPUT_REDUCER=cassandra
-
-#echo $CLASSPATH
-"$JAVA" -Xmx1G -ea -cp "$CLASSPATH" WordCount output_reducer=$OUTPUT_REDUCER
http://git-wip-us.apache.org/repos/asf/cassandra/blob/446e2537/examples/hadoop_word_count/bin/word_count_counters
----------------------------------------------------------------------
diff --git a/examples/hadoop_word_count/bin/word_count_counters b/examples/hadoop_word_count/bin/word_count_counters
deleted file mode 100755
index 122565d..0000000
--- a/examples/hadoop_word_count/bin/word_count_counters
+++ /dev/null
@@ -1,59 +0,0 @@
-#!/bin/sh
-
-# 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.
-
-cwd=`dirname $0`
-
-# Cassandra class files.
-if [ ! -d $cwd/../../../build/classes/main ]; then
- echo "Unable to locate cassandra class files" >&2
- exit 1
-fi
-
-# word_count Jar.
-if [ ! -e $cwd/../build/word_count.jar ]; then
- echo "Unable to locate word_count jar" >&2
- exit 1
-fi
-
-CLASSPATH=$CLASSPATH:$cwd/../conf
-CLASSPATH=$CLASSPATH:$cwd/../build/word_count.jar
-CLASSPATH=$CLASSPATH:$cwd/../../../build/classes/main
-CLASSPATH=$CLASSPATH:$cwd/../../../build/classes/thrift
-for jar in $cwd/../build/lib/jars/*.jar; do
- CLASSPATH=$CLASSPATH:$jar
-done
-for jar in $cwd/../../../lib/*.jar; do
- CLASSPATH=$CLASSPATH:$jar
-done
-for jar in $cwd/../../../build/lib/jars/*.jar; do
- CLASSPATH=$CLASSPATH:$jar
-done
-
-if [ -x $JAVA_HOME/bin/java ]; then
- JAVA=$JAVA_HOME/bin/java
-else
- JAVA=`which java`
-fi
-
-if [ "x$JAVA" = "x" ]; then
- echo "Java executable not found (hint: set JAVA_HOME)" >&2
- exit 1
-fi
-
-#echo $CLASSPATH
-"$JAVA" -Xmx1G -ea -cp "$CLASSPATH" WordCountCounters
http://git-wip-us.apache.org/repos/asf/cassandra/blob/446e2537/examples/hadoop_word_count/bin/word_count_setup
----------------------------------------------------------------------
diff --git a/examples/hadoop_word_count/bin/word_count_setup b/examples/hadoop_word_count/bin/word_count_setup
deleted file mode 100755
index 6e5650f..0000000
--- a/examples/hadoop_word_count/bin/word_count_setup
+++ /dev/null
@@ -1,61 +0,0 @@
-#!/bin/sh
-
-# 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.
-
-cwd=`dirname $0`
-
-# Cassandra class files.
-if [ ! -d $cwd/../../../build/classes/main ]; then
- echo "Unable to locate cassandra class files" >&2
- exit 1
-fi
-
-# word_count Jar.
-if [ ! -e $cwd/../build/word_count.jar ]; then
- echo "Unable to locate word_count jar" >&2
- exit 1
-fi
-
-CLASSPATH=$CLASSPATH:$cwd/../build/word_count.jar
-CLASSPATH=$CLASSPATH:.:$cwd/../../../build/classes/main
-CLASSPATH=$CLASSPATH:.:$cwd/../../../build/classes/thrift
-for jar in $cwd/../build/lib/jars/*.jar; do
- CLASSPATH=$CLASSPATH:$jar
-done
-for jar in $cwd/../../../lib/*.jar; do
- CLASSPATH=$CLASSPATH:$jar
-done
-for jar in $cwd/../../../build/lib/jars/*.jar; do
- CLASSPATH=$CLASSPATH:$jar
-done
-
-if [ -x $JAVA_HOME/bin/java ]; then
- JAVA=$JAVA_HOME/bin/java
-else
- JAVA=`which java`
-fi
-
-if [ "x$JAVA" = "x" ]; then
- echo "Java executable not found (hint: set JAVA_HOME)" >&2
- exit 1
-fi
-
-HOST=localhost
-PORT=9160
-FRAMED=true
-
-"$JAVA" -Xmx1G -ea -Dcassandra.host=$HOST -Dcassandra.port=$PORT -Dcassandra.framed=$FRAMED -cp "$CLASSPATH" WordCountSetup
http://git-wip-us.apache.org/repos/asf/cassandra/blob/446e2537/examples/hadoop_word_count/build.xml
----------------------------------------------------------------------
diff --git a/examples/hadoop_word_count/build.xml b/examples/hadoop_word_count/build.xml
deleted file mode 100644
index 939e1b3..0000000
--- a/examples/hadoop_word_count/build.xml
+++ /dev/null
@@ -1,113 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<!--
- ~ 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.
- -->
-<project default="jar" name="word_count" xmlns:ivy="antlib:org.apache.ivy.ant">
- <property name="cassandra.dir" value="../.." />
- <property name="cassandra.dir.lib" value="${cassandra.dir}/lib" />
- <property name="cassandra.classes" value="${cassandra.dir}/build/classes" />
- <property name="build.src" value="${basedir}/src" />
- <property name="build.dir" value="${basedir}/build" />
- <property name="ivy.lib.dir" value="${build.dir}/lib" />
- <property name="build.classes" value="${build.dir}/classes" />
- <property name="final.name" value="word_count" />
- <property name="ivy.version" value="2.1.0" />
- <property name="ivy.url"
- value="http://repo2.maven.org/maven2/org/apache/ivy/ivy" />
-
- <condition property="ivy.jar.exists">
- <available file="${build.dir}/ivy-${ivy.version}.jar" />
- </condition>
-
- <path id="autoivy.classpath">
- <fileset dir="${ivy.lib.dir}">
- <include name="**/*.jar" />
- </fileset>
- <pathelement location="${build.dir}/ivy-${ivy.version}.jar"/>
- </path>
-
- <path id="wordcount.build.classpath">
- <fileset dir="${ivy.lib.dir}">
- <include name="**/*.jar" />
- </fileset>
- <!-- cassandra dependencies -->
- <fileset dir="${cassandra.dir.lib}">
- <include name="**/*.jar" />
- </fileset>
- <fileset dir="${cassandra.dir}/build/lib/jars">
- <include name="**/*.jar" />
- </fileset>
- <pathelement location="${cassandra.classes}/main" />
- <pathelement location="${cassandra.classes}/thrift" />
- </path>
-
- <target name="init">
- <mkdir dir="${build.classes}" />
- </target>
-
- <target depends="init,ivy-retrieve-build" name="build">
- <javac destdir="${build.classes}">
- <src path="${build.src}" />
- <classpath refid="wordcount.build.classpath" />
- </javac>
- </target>
-
- <target name="jar" depends="build">
- <mkdir dir="${build.classes}/META-INF" />
- <jar jarfile="${build.dir}/${final.name}.jar">
- <fileset dir="${build.classes}" />
- <fileset dir="${cassandra.classes}/main" />
- <fileset dir="${cassandra.classes}/thrift" />
- <fileset dir="${cassandra.dir}">
- <include name="lib/**/*.jar" />
- </fileset>
- <zipfileset dir="${cassandra.dir}/build/lib/jars/" prefix="lib">
- <include name="**/*.jar" />
- </zipfileset>
- <fileset file="${basedir}/cassandra.yaml" />
- </jar>
- </target>
-
- <target name="clean">
- <delete dir="${build.dir}" />
- </target>
-
- <!--
- Ivy Specific targets
- to fetch Ivy and this project's dependencies
- -->
- <target name="ivy-download" unless="ivy.jar.exists">
- <echo>Downloading Ivy...</echo>
- <mkdir dir="${build.dir}" />
- <get src="${ivy.url}/${ivy.version}/ivy-${ivy.version}.jar"
- dest="${build.dir}/ivy-${ivy.version}.jar" usetimestamp="true" />
- </target>
-
- <target name="ivy-init" depends="ivy-download" unless="ivy.initialized">
- <mkdir dir="${ivy.lib.dir}"/>
- <taskdef resource="org/apache/ivy/ant/antlib.xml"
- uri="antlib:org.apache.ivy.ant"
- classpathref="autoivy.classpath"/>
- <property name="ivy.initialized" value="true"/>
- </target>
-
- <target name="ivy-retrieve-build" depends="ivy-init">
- <ivy:retrieve type="jar,source" sync="true"
- pattern="${ivy.lib.dir}/[type]s/[artifact]-[revision].[ext]" />
- </target>
-</project>
http://git-wip-us.apache.org/repos/asf/cassandra/blob/446e2537/examples/hadoop_word_count/conf/logback.xml
----------------------------------------------------------------------
diff --git a/examples/hadoop_word_count/conf/logback.xml b/examples/hadoop_word_count/conf/logback.xml
deleted file mode 100644
index 443bd1c..0000000
--- a/examples/hadoop_word_count/conf/logback.xml
+++ /dev/null
@@ -1,42 +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.
--->
-
-<configuration scan="true">
-
- <jmxConfigurator />
-
- <appender name="FILE" class="ch.qos.logback.core.FileAppender">
- <file>wc.out</file>
- <encoder>
- <pattern>%-5level [%thread] %date{ISO8601} %F:%L - %msg%n</pattern>
- </encoder>
- </appender>
-
- <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
- <encoder>
- <pattern>%-5level %date{HH:mm:ss,SSS} %msg%n</pattern>
- </encoder>
- </appender>
-
- <root level="INFO">
- <appender-ref ref="FILE" />
- <appender-ref ref="STDOUT" />
- </root>
-
-</configuration>
http://git-wip-us.apache.org/repos/asf/cassandra/blob/446e2537/examples/hadoop_word_count/ivy.xml
----------------------------------------------------------------------
diff --git a/examples/hadoop_word_count/ivy.xml b/examples/hadoop_word_count/ivy.xml
deleted file mode 100644
index 2016eb8..0000000
--- a/examples/hadoop_word_count/ivy.xml
+++ /dev/null
@@ -1,24 +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.
- -->
-<ivy-module version="2.0">
- <info organisation="apache-cassandra" module="word-count"/>
- <dependencies>
- <dependency org="org.apache.hadoop" name="hadoop-core" rev="1.0.3"/>
- </dependencies>
-</ivy-module>
http://git-wip-us.apache.org/repos/asf/cassandra/blob/446e2537/examples/hadoop_word_count/src/WordCount.java
----------------------------------------------------------------------
diff --git a/examples/hadoop_word_count/src/WordCount.java b/examples/hadoop_word_count/src/WordCount.java
deleted file mode 100644
index d092f1f..0000000
--- a/examples/hadoop_word_count/src/WordCount.java
+++ /dev/null
@@ -1,222 +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.
- */
-
-import java.io.IOException;
-import java.nio.ByteBuffer;
-import java.util.*;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import org.apache.cassandra.hadoop.*;
-import org.apache.cassandra.thrift.*;
-import org.apache.cassandra.utils.ByteBufferUtil;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.conf.Configured;
-import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.io.IntWritable;
-import org.apache.hadoop.io.Text;
-import org.apache.hadoop.mapreduce.Job;
-import org.apache.hadoop.mapreduce.Mapper;
-import org.apache.hadoop.mapreduce.Reducer;
-import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
-import org.apache.hadoop.util.Tool;
-import org.apache.hadoop.util.ToolRunner;
-
-/**
- * This counts the occurrences of words in ColumnFamily Standard1, that has a single column (that we care about)
- * "text" containing a sequence of words.
- *
- * For each word, we output the total number of occurrences across all texts.
- *
- * When outputting to Cassandra, we write the word counts as a {word, count} column/value pair,
- * with a row key equal to the name of the source column we read the words from.
- */
-public class WordCount extends Configured implements Tool
-{
- private static final Logger logger = LoggerFactory.getLogger(WordCount.class);
-
- static final String KEYSPACE = "wordcount";
- static final String COLUMN_FAMILY = "input_words";
-
- static final String OUTPUT_REDUCER_VAR = "output_reducer";
- static final String OUTPUT_COLUMN_FAMILY = "output_words";
- private static final String OUTPUT_PATH_PREFIX = "/tmp/word_count";
-
- private static final String CONF_COLUMN_NAME = "columnname";
-
- public static void main(String[] args) throws Exception
- {
- // Let ToolRunner handle generic command-line options
- ToolRunner.run(new Configuration(), new WordCount(), args);
- System.exit(0);
- }
-
- public static class TokenizerMapper extends Mapper<ByteBuffer, SortedMap<ByteBuffer, ColumnFamilyRecordReader.Column>, Text, IntWritable>
- {
- private final static IntWritable one = new IntWritable(1);
- private Text word = new Text();
- private ByteBuffer sourceColumn;
-
- protected void setup(org.apache.hadoop.mapreduce.Mapper.Context context)
- throws IOException, InterruptedException
- {
- }
-
- public void map(ByteBuffer key, SortedMap<ByteBuffer, ColumnFamilyRecordReader.Column> columns, Context context) throws IOException, InterruptedException
- {
- for (ColumnFamilyRecordReader.Column column : columns.values())
- {
- String name = ByteBufferUtil.string(column.name);
- String value = null;
-
- if (name.contains("int"))
- value = String.valueOf(ByteBufferUtil.toInt(column.value));
- else
- value = ByteBufferUtil.string(column.value);
-
- logger.debug("read {}:{}={} from {}",
- new Object[] {ByteBufferUtil.string(key), name, value, context.getInputSplit()});
-
- StringTokenizer itr = new StringTokenizer(value);
- while (itr.hasMoreTokens())
- {
- word.set(itr.nextToken());
- context.write(word, one);
- }
- }
- }
- }
-
- public static class ReducerToFilesystem extends Reducer<Text, IntWritable, Text, IntWritable>
- {
- public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException
- {
- int sum = 0;
- for (IntWritable val : values)
- sum += val.get();
- context.write(key, new IntWritable(sum));
- }
- }
-
- public static class ReducerToCassandra extends Reducer<Text, IntWritable, ByteBuffer, List<Mutation>>
- {
- private ByteBuffer outputKey;
-
- protected void setup(org.apache.hadoop.mapreduce.Reducer.Context context)
- throws IOException, InterruptedException
- {
- outputKey = ByteBufferUtil.bytes(context.getConfiguration().get(CONF_COLUMN_NAME));
- }
-
- public void reduce(Text word, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException
- {
- int sum = 0;
- for (IntWritable val : values)
- sum += val.get();
- context.write(outputKey, Collections.singletonList(getMutation(word, sum)));
- }
-
- private static Mutation getMutation(Text word, int sum)
- {
- org.apache.cassandra.thrift.Column c = new org.apache.cassandra.thrift.Column();
- c.setName(Arrays.copyOf(word.getBytes(), word.getLength()));
- c.setValue(ByteBufferUtil.bytes(sum));
- c.setTimestamp(System.currentTimeMillis());
-
- Mutation m = new Mutation();
- m.setColumn_or_supercolumn(new ColumnOrSuperColumn());
- m.column_or_supercolumn.setColumn(c);
- return m;
- }
- }
-
- public int run(String[] args) throws Exception
- {
- String outputReducerType = "filesystem";
- if (args != null && args[0].startsWith(OUTPUT_REDUCER_VAR))
- {
- String[] s = args[0].split("=");
- if (s != null && s.length == 2)
- outputReducerType = s[1];
- }
- logger.info("output reducer type: " + outputReducerType);
-
- // use a smaller page size that doesn't divide the row count evenly to exercise the paging logic better
- ConfigHelper.setRangeBatchSize(getConf(), 99);
-
- for (int i = 0; i < WordCountSetup.TEST_COUNT; i++)
- {
- String columnName = "text" + i;
-
- Job job = new Job(getConf(), "wordcount");
- job.setJarByClass(WordCount.class);
- job.setMapperClass(TokenizerMapper.class);
-
- if (outputReducerType.equalsIgnoreCase("filesystem"))
- {
- job.setCombinerClass(ReducerToFilesystem.class);
- job.setReducerClass(ReducerToFilesystem.class);
- job.setOutputKeyClass(Text.class);
- job.setOutputValueClass(IntWritable.class);
- FileOutputFormat.setOutputPath(job, new Path(OUTPUT_PATH_PREFIX + i));
- }
- else
- {
- job.setReducerClass(ReducerToCassandra.class);
-
- job.setMapOutputKeyClass(Text.class);
- job.setMapOutputValueClass(IntWritable.class);
- job.setOutputKeyClass(ByteBuffer.class);
- job.setOutputValueClass(List.class);
-
- job.setOutputFormatClass(ColumnFamilyOutputFormat.class);
-
- ConfigHelper.setOutputColumnFamily(job.getConfiguration(), KEYSPACE, OUTPUT_COLUMN_FAMILY);
- job.getConfiguration().set(CONF_COLUMN_NAME, "sum");
- }
-
- job.setInputFormatClass(ColumnFamilyInputFormat.class);
-
- ConfigHelper.setInputRpcPort(job.getConfiguration(), "9160");
- ConfigHelper.setInputInitialAddress(job.getConfiguration(), "localhost");
- ConfigHelper.setInputPartitioner(job.getConfiguration(), "Murmur3Partitioner");
- ConfigHelper.setInputColumnFamily(job.getConfiguration(), KEYSPACE, COLUMN_FAMILY);
- SlicePredicate predicate = new SlicePredicate().setColumn_names(Arrays.asList(ByteBufferUtil.bytes(columnName)));
- ConfigHelper.setInputSlicePredicate(job.getConfiguration(), predicate);
-
- if (i == 4)
- {
- IndexExpression expr = new IndexExpression(ByteBufferUtil.bytes("int4"), IndexOperator.EQ, ByteBufferUtil.bytes(0));
- ConfigHelper.setInputRange(job.getConfiguration(), Arrays.asList(expr));
- }
-
- if (i == 5)
- {
- // this will cause the predicate to be ignored in favor of scanning everything as a wide row
- ConfigHelper.setInputColumnFamily(job.getConfiguration(), KEYSPACE, COLUMN_FAMILY, true);
- }
-
- ConfigHelper.setOutputInitialAddress(job.getConfiguration(), "localhost");
- ConfigHelper.setOutputPartitioner(job.getConfiguration(), "Murmur3Partitioner");
-
- job.waitForCompletion(true);
- }
- return 0;
- }
-}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/446e2537/examples/hadoop_word_count/src/WordCountCounters.java
----------------------------------------------------------------------
diff --git a/examples/hadoop_word_count/src/WordCountCounters.java b/examples/hadoop_word_count/src/WordCountCounters.java
deleted file mode 100644
index 98c8579..0000000
--- a/examples/hadoop_word_count/src/WordCountCounters.java
+++ /dev/null
@@ -1,104 +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.
- */
-
-import java.io.IOException;
-import java.nio.ByteBuffer;
-import java.util.SortedMap;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import org.apache.cassandra.hadoop.ColumnFamilyInputFormat;
-import org.apache.cassandra.hadoop.ColumnFamilyRecordReader;
-import org.apache.cassandra.hadoop.ConfigHelper;
-import org.apache.cassandra.thrift.SlicePredicate;
-import org.apache.cassandra.thrift.SliceRange;
-import org.apache.cassandra.utils.ByteBufferUtil;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.conf.Configured;
-import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.io.LongWritable;
-import org.apache.hadoop.io.Text;
-import org.apache.hadoop.mapreduce.Job;
-import org.apache.hadoop.mapreduce.Mapper;
-import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
-import org.apache.hadoop.util.Tool;
-import org.apache.hadoop.util.ToolRunner;
-
-/**
- * This sums the word count stored in the input_words_count ColumnFamily for the key "key-if-verse1".
- *
- * Output is written to a text file.
- */
-public class WordCountCounters extends Configured implements Tool
-{
- private static final Logger logger = LoggerFactory.getLogger(WordCountCounters.class);
-
- static final String COUNTER_COLUMN_FAMILY = "input_words_count";
- private static final String OUTPUT_PATH_PREFIX = "/tmp/word_count_counters";
-
-
- public static void main(String[] args) throws Exception
- {
- // Let ToolRunner handle generic command-line options
- ToolRunner.run(new Configuration(), new WordCountCounters(), args);
- System.exit(0);
- }
-
- public static class SumMapper extends Mapper<ByteBuffer, SortedMap<ByteBuffer, ColumnFamilyRecordReader.Column>, Text, LongWritable>
- {
- public void map(ByteBuffer key, SortedMap<ByteBuffer, ColumnFamilyRecordReader.Column> columns, Context context) throws IOException, InterruptedException
- {
- long sum = 0;
- for (ColumnFamilyRecordReader.Column column : columns.values())
- {
- logger.debug("read " + key + ":" + ByteBufferUtil.string(column.name) + " from " + context.getInputSplit());
- sum += ByteBufferUtil.toLong(column.value);
- }
- context.write(new Text(ByteBufferUtil.string(key)), new LongWritable(sum));
- }
- }
-
- public int run(String[] args) throws Exception
- {
- Job job = new Job(getConf(), "wordcountcounters");
- job.setJarByClass(WordCountCounters.class);
- job.setMapperClass(SumMapper.class);
-
- job.setOutputKeyClass(Text.class);
- job.setOutputValueClass(LongWritable.class);
- FileOutputFormat.setOutputPath(job, new Path(OUTPUT_PATH_PREFIX));
-
-
- job.setInputFormatClass(ColumnFamilyInputFormat.class);
-
- ConfigHelper.setInputRpcPort(job.getConfiguration(), "9160");
- ConfigHelper.setInputInitialAddress(job.getConfiguration(), "localhost");
- ConfigHelper.setInputPartitioner(job.getConfiguration(), "org.apache.cassandra.dht.Murmur3Partitioner");
- ConfigHelper.setInputColumnFamily(job.getConfiguration(), WordCount.KEYSPACE, WordCountCounters.COUNTER_COLUMN_FAMILY);
- SlicePredicate predicate = new SlicePredicate().setSlice_range(
- new SliceRange().
- setStart(ByteBufferUtil.EMPTY_BYTE_BUFFER).
- setFinish(ByteBufferUtil.EMPTY_BYTE_BUFFER).
- setCount(100));
- ConfigHelper.setInputSlicePredicate(job.getConfiguration(), predicate);
-
- job.waitForCompletion(true);
- return 0;
- }
-}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/446e2537/examples/hadoop_word_count/src/WordCountSetup.java
----------------------------------------------------------------------
diff --git a/examples/hadoop_word_count/src/WordCountSetup.java b/examples/hadoop_word_count/src/WordCountSetup.java
deleted file mode 100644
index 0ef5341..0000000
--- a/examples/hadoop_word_count/src/WordCountSetup.java
+++ /dev/null
@@ -1,239 +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.
- */
-
-import java.nio.ByteBuffer;
-import java.util.*;
-import java.util.concurrent.TimeUnit;
-
-import org.apache.cassandra.thrift.*;
-import org.apache.cassandra.utils.ByteBufferUtil;
-import org.apache.thrift.TException;
-import org.apache.thrift.protocol.TBinaryProtocol;
-import org.apache.thrift.protocol.TProtocol;
-import org.apache.thrift.transport.TFramedTransport;
-import org.apache.thrift.transport.TSocket;
-import org.apache.thrift.transport.TTransport;
-import org.apache.thrift.transport.TTransportException;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.google.common.util.concurrent.Uninterruptibles;
-
-public class WordCountSetup
-{
- private static final Logger logger = LoggerFactory.getLogger(WordCountSetup.class);
-
- public static final int TEST_COUNT = 6;
-
- public static void main(String[] args) throws Exception
- {
- Cassandra.Iface client = createConnection();
-
- setupKeyspace(client);
-
- client.set_keyspace(WordCount.KEYSPACE);
-
- Map<ByteBuffer, Map<String, List<Mutation>>> mutationMap;
- Column c;
-
- // text0: no rows
-
- // text1: 1 row, 1 word
- c = new Column()
- .setName(ByteBufferUtil.bytes("text1"))
- .setValue(ByteBufferUtil.bytes("word1"))
- .setTimestamp(System.currentTimeMillis());
- mutationMap = getMutationMap(ByteBufferUtil.bytes("key0"), WordCount.COLUMN_FAMILY, c);
- client.batch_mutate(mutationMap, ConsistencyLevel.ONE);
- logger.info("added text1");
-
- // text1: 1 row, 2 word
- c = new Column()
- .setName(ByteBufferUtil.bytes("text2"))
- .setValue(ByteBufferUtil.bytes("word1 word2"))
- .setTimestamp(System.currentTimeMillis());
- mutationMap = getMutationMap(ByteBufferUtil.bytes("key0"), WordCount.COLUMN_FAMILY, c);
- client.batch_mutate(mutationMap, ConsistencyLevel.ONE);
- logger.info("added text2");
-
- // text3: 1000 rows, 1 word
- mutationMap = new HashMap<ByteBuffer, Map<String, List<Mutation>>>();
- for (int i = 0; i < 1000; i++)
- {
- c = new Column()
- .setName(ByteBufferUtil.bytes("text3"))
- .setValue(ByteBufferUtil.bytes("word1"))
- .setTimestamp(System.currentTimeMillis());
- addToMutationMap(mutationMap, ByteBufferUtil.bytes("key" + i), WordCount.COLUMN_FAMILY, c);
- }
- client.batch_mutate(mutationMap, ConsistencyLevel.ONE);
- logger.info("added text3");
-
- // text4: 1000 rows, 1 word, one column to filter on
- mutationMap = new HashMap<ByteBuffer, Map<String, List<Mutation>>>();
- for (int i = 0; i < 1000; i++)
- {
- Column c1 = new Column()
- .setName(ByteBufferUtil.bytes("text4"))
- .setValue(ByteBufferUtil.bytes("word1"))
- .setTimestamp(System.currentTimeMillis());
- Column c2 = new Column()
- .setName(ByteBufferUtil.bytes("int4"))
- .setValue(ByteBufferUtil.bytes(i % 4))
- .setTimestamp(System.currentTimeMillis());
- ByteBuffer key = ByteBufferUtil.bytes("key" + i);
- addToMutationMap(mutationMap, key, WordCount.COLUMN_FAMILY, c1);
- addToMutationMap(mutationMap, key, WordCount.COLUMN_FAMILY, c2);
- }
- client.batch_mutate(mutationMap, ConsistencyLevel.ONE);
- logger.info("added text4");
-
- // sentence data for the counters
- final ByteBuffer key = ByteBufferUtil.bytes("key-if-verse1");
- final ColumnParent colParent = new ColumnParent(WordCountCounters.COUNTER_COLUMN_FAMILY);
- for (String sentence : sentenceData())
- {
- client.add(key,
- colParent,
- new CounterColumn(ByteBufferUtil.bytes(sentence),
- (long) sentence.split("\\s").length),
- ConsistencyLevel.ONE);
- }
- logger.info("added key-if-verse1");
-
- System.exit(0);
- }
-
- private static Map<ByteBuffer, Map<String, List<Mutation>>> getMutationMap(ByteBuffer key, String cf, Column c)
- {
- Map<ByteBuffer, Map<String, List<Mutation>>> mutationMap = new HashMap<ByteBuffer, Map<String, List<Mutation>>>();
- addToMutationMap(mutationMap, key, cf, c);
- return mutationMap;
- }
-
- private static void addToMutationMap(Map<ByteBuffer, Map<String, List<Mutation>>> mutationMap, ByteBuffer key, String cf, Column c)
- {
- Map<String, List<Mutation>> cfMutation = mutationMap.get(key);
- if (cfMutation == null)
- {
- cfMutation = new HashMap<String, List<Mutation>>();
- mutationMap.put(key, cfMutation);
- }
-
- List<Mutation> mutationList = cfMutation.get(cf);
- if (mutationList == null)
- {
- mutationList = new ArrayList<Mutation>();
- cfMutation.put(cf, mutationList);
- }
-
- ColumnOrSuperColumn cc = new ColumnOrSuperColumn();
- Mutation m = new Mutation();
-
- cc.setColumn(c);
- m.setColumn_or_supercolumn(cc);
- mutationList.add(m);
- }
-
- private static void setupKeyspace(Cassandra.Iface client) throws TException, InvalidRequestException, SchemaDisagreementException
- {
- List<CfDef> cfDefList = new ArrayList<CfDef>();
- CfDef input = new CfDef(WordCount.KEYSPACE, WordCount.COLUMN_FAMILY);
- input.setComparator_type("AsciiType");
- input.setColumn_metadata(Arrays.asList(new ColumnDef(ByteBufferUtil.bytes("text1"), "AsciiType"),
- new ColumnDef(ByteBufferUtil.bytes("text2"), "AsciiType"),
- new ColumnDef(ByteBufferUtil.bytes("text3"), "AsciiType"),
- new ColumnDef(ByteBufferUtil.bytes("text4"), "AsciiType"),
- new ColumnDef(ByteBufferUtil.bytes("int4"), "Int32Type").setIndex_name("int4idx").setIndex_type(IndexType.KEYS)));
- cfDefList.add(input);
-
- CfDef output = new CfDef(WordCount.KEYSPACE, WordCount.OUTPUT_COLUMN_FAMILY);
- output.setComparator_type("AsciiType");
- output.setDefault_validation_class("Int32Type");
- cfDefList.add(output);
-
- CfDef counterInput = new CfDef(WordCount.KEYSPACE, WordCountCounters.COUNTER_COLUMN_FAMILY);
- counterInput.setComparator_type("UTF8Type");
- counterInput.setDefault_validation_class("CounterColumnType");
- cfDefList.add(counterInput);
-
- KsDef ksDef = new KsDef(WordCount.KEYSPACE, "org.apache.cassandra.locator.SimpleStrategy", cfDefList);
- ksDef.putToStrategy_options("replication_factor", "1");
- client.system_add_keyspace(ksDef);
-
- int magnitude = getNumberOfHosts(client);
- Uninterruptibles.sleepUninterruptibly(magnitude, TimeUnit.SECONDS);
- }
-
- private static int getNumberOfHosts(Cassandra.Iface client)
- throws InvalidRequestException, UnavailableException, TimedOutException, TException
- {
- client.set_keyspace("system");
- SlicePredicate predicate = new SlicePredicate();
- SliceRange sliceRange = new SliceRange();
- sliceRange.setStart(new byte[0]);
- sliceRange.setFinish(new byte[0]);
- predicate.setSlice_range(sliceRange);
-
- KeyRange keyrRange = new KeyRange();
- keyrRange.setStart_key(new byte[0]);
- keyrRange.setEnd_key(new byte[0]);
- //keyrRange.setCount(100);
-
- ColumnParent parent = new ColumnParent("peers");
-
- List<KeySlice> ls = client.get_range_slices(parent, predicate, keyrRange, ConsistencyLevel.ONE);
-
- return ls.size();
- }
-
- private static Cassandra.Iface createConnection() throws TTransportException
- {
- if (System.getProperty("cassandra.host") == null || System.getProperty("cassandra.port") == null)
- {
- logger.warn("cassandra.host or cassandra.port is not defined, using default");
- }
- return createConnection(System.getProperty("cassandra.host", "localhost"),
- Integer.valueOf(System.getProperty("cassandra.port", "9160")));
- }
-
- private static Cassandra.Client createConnection(String host, Integer port) throws TTransportException
- {
- TSocket socket = new TSocket(host, port);
- TTransport trans = new TFramedTransport(socket);
- trans.open();
- TProtocol protocol = new TBinaryProtocol(trans);
-
- return new Cassandra.Client(protocol);
- }
-
- private static String[] sentenceData()
- { // Public domain context, source http://en.wikisource.org/wiki/If%E2%80%94
- return new String[]{
- "If you can keep your head when all about you",
- "Are losing theirs and blaming it on you",
- "If you can trust yourself when all men doubt you,",
- "But make allowance for their doubting too:",
- "If you can wait and not be tired by waiting,",
- "Or being lied about, don’t deal in lies,",
- "Or being hated, don’t give way to hating,",
- "And yet don’t look too good, nor talk too wise;"
- };
- }
-}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/446e2537/src/java/org/apache/cassandra/hadoop/AbstractColumnFamilyInputFormat.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/hadoop/AbstractColumnFamilyInputFormat.java b/src/java/org/apache/cassandra/hadoop/AbstractColumnFamilyInputFormat.java
deleted file mode 100644
index 4dd53ff..0000000
--- a/src/java/org/apache/cassandra/hadoop/AbstractColumnFamilyInputFormat.java
+++ /dev/null
@@ -1,296 +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.cassandra.hadoop;
-
-import java.io.IOException;
-import java.util.*;
-import java.util.concurrent.*;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.datastax.driver.core.Host;
-import com.datastax.driver.core.Metadata;
-import com.datastax.driver.core.ResultSet;
-import com.datastax.driver.core.Row;
-import com.datastax.driver.core.Session;
-import com.datastax.driver.core.TokenRange;
-import org.apache.cassandra.db.SystemKeyspace;
-import org.apache.cassandra.dht.ByteOrderedPartitioner;
-import org.apache.cassandra.dht.IPartitioner;
-import org.apache.cassandra.dht.OrderPreservingPartitioner;
-import org.apache.cassandra.dht.Range;
-import org.apache.cassandra.dht.Token;
-import org.apache.cassandra.hadoop.cql3.*;
-import org.apache.cassandra.thrift.KeyRange;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.mapred.JobConf;
-import org.apache.hadoop.mapreduce.*;
-
-public abstract class AbstractColumnFamilyInputFormat<K, Y> extends InputFormat<K, Y> implements org.apache.hadoop.mapred.InputFormat<K, Y>
-{
- private static final Logger logger = LoggerFactory.getLogger(AbstractColumnFamilyInputFormat.class);
-
- public static final String MAPRED_TASK_ID = "mapred.task.id";
- // The simple fact that we need this is because the old Hadoop API wants us to "write"
- // to the key and value whereas the new asks for it.
- // I choose 8kb as the default max key size (instantiated only once), but you can
- // override it in your jobConf with this setting.
- public static final String CASSANDRA_HADOOP_MAX_KEY_SIZE = "cassandra.hadoop.max_key_size";
- public static final int CASSANDRA_HADOOP_MAX_KEY_SIZE_DEFAULT = 8192;
-
- private String keyspace;
- private String cfName;
- private IPartitioner partitioner;
- private Session session;
-
- protected void validateConfiguration(Configuration conf)
- {
- if (ConfigHelper.getInputKeyspace(conf) == null || ConfigHelper.getInputColumnFamily(conf) == null)
- {
- throw new UnsupportedOperationException("you must set the keyspace and table with setInputColumnFamily()");
- }
- if (ConfigHelper.getInputInitialAddress(conf) == null)
- throw new UnsupportedOperationException("You must set the initial output address to a Cassandra node with setInputInitialAddress");
- if (ConfigHelper.getInputPartitioner(conf) == null)
- throw new UnsupportedOperationException("You must set the Cassandra partitioner class with setInputPartitioner");
- }
-
- public List<InputSplit> getSplits(JobContext context) throws IOException
- {
- Configuration conf = HadoopCompat.getConfiguration(context);
-
- validateConfiguration(conf);
-
- keyspace = ConfigHelper.getInputKeyspace(conf);
- cfName = ConfigHelper.getInputColumnFamily(conf);
- partitioner = ConfigHelper.getInputPartitioner(conf);
- logger.debug("partitioner is {}", partitioner);
-
- // canonical ranges and nodes holding replicas
- Map<TokenRange, Set<Host>> masterRangeNodes = getRangeMap(conf, keyspace);
-
- // canonical ranges, split into pieces, fetching the splits in parallel
- ExecutorService executor = new ThreadPoolExecutor(0, 128, 60L, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>());
- List<InputSplit> splits = new ArrayList<>();
-
- try
- {
- List<Future<List<InputSplit>>> splitfutures = new ArrayList<>();
- KeyRange jobKeyRange = ConfigHelper.getInputKeyRange(conf);
- Range<Token> jobRange = null;
- if (jobKeyRange != null)
- {
- if (jobKeyRange.start_key != null)
- {
- if (!partitioner.preservesOrder())
- throw new UnsupportedOperationException("KeyRange based on keys can only be used with a order preserving partitioner");
- if (jobKeyRange.start_token != null)
- throw new IllegalArgumentException("only start_key supported");
- if (jobKeyRange.end_token != null)
- throw new IllegalArgumentException("only start_key supported");
- jobRange = new Range<>(partitioner.getToken(jobKeyRange.start_key),
- partitioner.getToken(jobKeyRange.end_key));
- }
- else if (jobKeyRange.start_token != null)
- {
- jobRange = new Range<>(partitioner.getTokenFactory().fromString(jobKeyRange.start_token),
- partitioner.getTokenFactory().fromString(jobKeyRange.end_token));
- }
- else
- {
- logger.warn("ignoring jobKeyRange specified without start_key or start_token");
- }
- }
-
- session = CqlConfigHelper.getInputCluster(ConfigHelper.getInputInitialAddress(conf).split(","), conf).connect();
- Metadata metadata = session.getCluster().getMetadata();
-
- for (TokenRange range : masterRangeNodes.keySet())
- {
- if (jobRange == null)
- {
- // for each tokenRange, pick a live owner and ask it to compute bite-sized splits
- splitfutures.add(executor.submit(new SplitCallable(range, masterRangeNodes.get(range), conf)));
- }
- else
- {
- TokenRange jobTokenRange = rangeToTokenRange(metadata, jobRange);
- if (range.intersects(jobTokenRange))
- {
- for (TokenRange intersection: range.intersectWith(jobTokenRange))
- {
- // for each tokenRange, pick a live owner and ask it to compute bite-sized splits
- splitfutures.add(executor.submit(new SplitCallable(intersection, masterRangeNodes.get(range), conf)));
- }
- }
- }
- }
-
- // wait until we have all the results back
- for (Future<List<InputSplit>> futureInputSplits : splitfutures)
- {
- try
- {
- splits.addAll(futureInputSplits.get());
- }
- catch (Exception e)
- {
- throw new IOException("Could not get input splits", e);
- }
- }
- }
- finally
- {
- executor.shutdownNow();
- }
-
- assert splits.size() > 0;
- Collections.shuffle(splits, new Random(System.nanoTime()));
- return splits;
- }
-
- private TokenRange rangeToTokenRange(Metadata metadata, Range<Token> range)
- {
- return metadata.newTokenRange(metadata.newToken(partitioner.getTokenFactory().toString(range.left)),
- metadata.newToken(partitioner.getTokenFactory().toString(range.right)));
- }
-
- /**
- * Gets a token tokenRange and splits it up according to the suggested
- * size into input splits that Hadoop can use.
- */
- class SplitCallable implements Callable<List<InputSplit>>
- {
-
- private final TokenRange tokenRange;
- private final Set<Host> hosts;
- private final Configuration conf;
-
- public SplitCallable(TokenRange tr, Set<Host> hosts, Configuration conf)
- {
- this.tokenRange = tr;
- this.hosts = hosts;
- this.conf = conf;
- }
-
- public List<InputSplit> call() throws Exception
- {
- ArrayList<InputSplit> splits = new ArrayList<>();
- Map<TokenRange, Long> subSplits;
- subSplits = getSubSplits(keyspace, cfName, tokenRange, conf);
- // turn the sub-ranges into InputSplits
- String[] endpoints = new String[hosts.size()];
-
- // hadoop needs hostname, not ip
- int endpointIndex = 0;
- for (Host endpoint : hosts)
- endpoints[endpointIndex++] = endpoint.getAddress().getHostName();
-
- boolean partitionerIsOpp = partitioner instanceof OrderPreservingPartitioner || partitioner instanceof ByteOrderedPartitioner;
-
- for (TokenRange subSplit : subSplits.keySet())
- {
- List<TokenRange> ranges = subSplit.unwrap();
- for (TokenRange subrange : ranges)
- {
- ColumnFamilySplit split =
- new ColumnFamilySplit(
- partitionerIsOpp ?
- subrange.getStart().toString().substring(2) : subrange.getStart().toString(),
- partitionerIsOpp ?
- subrange.getEnd().toString().substring(2) : subrange.getStart().toString(),
- subSplits.get(subSplit),
- endpoints);
-
- logger.debug("adding {}", split);
- splits.add(split);
- }
- }
- return splits;
- }
- }
-
- private Map<TokenRange, Long> getSubSplits(String keyspace, String cfName, TokenRange range, Configuration conf) throws IOException
- {
- int splitSize = ConfigHelper.getInputSplitSize(conf);
- try
- {
- return describeSplits(keyspace, cfName, range, splitSize);
- }
- catch (Exception e)
- {
- throw new RuntimeException(e);
- }
- }
-
- private Map<TokenRange, Set<Host>> getRangeMap(Configuration conf, String keyspace)
- {
- try (Session session = CqlConfigHelper.getInputCluster(ConfigHelper.getInputInitialAddress(conf).split(","), conf).connect())
- {
- Map<TokenRange, Set<Host>> map = new HashMap<>();
- Metadata metadata = session.getCluster().getMetadata();
- for (TokenRange tokenRange : metadata.getTokenRanges())
- map.put(tokenRange, metadata.getReplicas('"' + keyspace + '"', tokenRange));
- return map;
- }
- }
-
- private Map<TokenRange, Long> describeSplits(String keyspace, String table, TokenRange tokenRange, int splitSize)
- {
- String query = String.format("SELECT mean_partition_size, partitions_count " +
- "FROM %s.%s " +
- "WHERE keyspace_name = ? AND table_name = ? AND range_start = ? AND range_end = ?",
- SystemKeyspace.NAME,
- SystemKeyspace.SIZE_ESTIMATES);
-
- ResultSet resultSet = session.execute(query, keyspace, table, tokenRange.getStart().toString(), tokenRange.getEnd().toString());
-
- Row row = resultSet.one();
- // If we have no data on this split, return the full split i.e., do not sub-split
- // Assume smallest granularity of partition count available from CASSANDRA-7688
- if (row == null)
- {
- Map<TokenRange, Long> wrappedTokenRange = new HashMap<>();
- wrappedTokenRange.put(tokenRange, (long) 128);
- return wrappedTokenRange;
- }
-
- long meanPartitionSize = row.getLong("mean_partition_size");
- long partitionCount = row.getLong("partitions_count");
-
- int splitCount = (int)((meanPartitionSize * partitionCount) / splitSize);
- List<TokenRange> splitRanges = tokenRange.splitEvenly(splitCount);
- Map<TokenRange, Long> rangesWithLength = new HashMap<>();
- for (TokenRange range : splitRanges)
- rangesWithLength.put(range, partitionCount/splitCount);
-
- return rangesWithLength;
- }
-
- // Old Hadoop API
- public org.apache.hadoop.mapred.InputSplit[] getSplits(JobConf jobConf, int numSplits) throws IOException
- {
- TaskAttemptContext tac = HadoopCompat.newTaskAttemptContext(jobConf, new TaskAttemptID());
- List<org.apache.hadoop.mapreduce.InputSplit> newInputSplits = this.getSplits(tac);
- org.apache.hadoop.mapred.InputSplit[] oldInputSplits = new org.apache.hadoop.mapred.InputSplit[newInputSplits.size()];
- for (int i = 0; i < newInputSplits.size(); i++)
- oldInputSplits[i] = (ColumnFamilySplit)newInputSplits.get(i);
- return oldInputSplits;
- }
-}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/446e2537/src/java/org/apache/cassandra/hadoop/BulkOutputFormat.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/hadoop/BulkOutputFormat.java b/src/java/org/apache/cassandra/hadoop/BulkOutputFormat.java
deleted file mode 100644
index 5282279..0000000
--- a/src/java/org/apache/cassandra/hadoop/BulkOutputFormat.java
+++ /dev/null
@@ -1,91 +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.cassandra.hadoop;
-
-
-import java.io.IOException;
-import java.nio.ByteBuffer;
-import java.util.List;
-
-import org.apache.cassandra.thrift.Mutation;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.mapreduce.*;
-
-@Deprecated
-public class BulkOutputFormat extends OutputFormat<ByteBuffer,List<Mutation>>
- implements org.apache.hadoop.mapred.OutputFormat<ByteBuffer,List<Mutation>>
-{
- /** Fills the deprecated OutputFormat interface for streaming. */
- @Deprecated
- public BulkRecordWriter getRecordWriter(org.apache.hadoop.fs.FileSystem filesystem, org.apache.hadoop.mapred.JobConf job, String name, org.apache.hadoop.util.Progressable progress) throws IOException
- {
- return new BulkRecordWriter(job, progress);
- }
-
- @Override
- public BulkRecordWriter getRecordWriter(final TaskAttemptContext context) throws IOException, InterruptedException
- {
- return new BulkRecordWriter(context);
- }
-
-
- @Override
- public void checkOutputSpecs(JobContext context)
- {
- checkOutputSpecs(HadoopCompat.getConfiguration(context));
- }
-
- private void checkOutputSpecs(Configuration conf)
- {
- if (ConfigHelper.getOutputKeyspace(conf) == null)
- {
- throw new UnsupportedOperationException("you must set the keyspace with setColumnFamily()");
- }
- }
-
- @Override
- public OutputCommitter getOutputCommitter(TaskAttemptContext context) throws IOException, InterruptedException
- {
- return new NullOutputCommitter();
- }
-
- /** Fills the deprecated OutputFormat interface for streaming. */
- @Deprecated
- public void checkOutputSpecs(org.apache.hadoop.fs.FileSystem filesystem, org.apache.hadoop.mapred.JobConf job) throws IOException
- {
- checkOutputSpecs(job);
- }
-
- public static class NullOutputCommitter extends OutputCommitter
- {
- public void abortTask(TaskAttemptContext taskContext) { }
-
- public void cleanupJob(JobContext jobContext) { }
-
- public void commitTask(TaskAttemptContext taskContext) { }
-
- public boolean needsTaskCommit(TaskAttemptContext taskContext)
- {
- return false;
- }
-
- public void setupJob(JobContext jobContext) { }
-
- public void setupTask(TaskAttemptContext taskContext) { }
- }
-}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/446e2537/src/java/org/apache/cassandra/hadoop/BulkRecordWriter.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/hadoop/BulkRecordWriter.java b/src/java/org/apache/cassandra/hadoop/BulkRecordWriter.java
deleted file mode 100644
index 99abf9f..0000000
--- a/src/java/org/apache/cassandra/hadoop/BulkRecordWriter.java
+++ /dev/null
@@ -1,296 +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.cassandra.hadoop;
-
-import java.io.Closeable;
-import java.io.File;
-import java.io.IOException;
-import java.net.InetAddress;
-import java.net.UnknownHostException;
-import java.nio.ByteBuffer;
-import java.util.*;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.Future;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import org.apache.cassandra.config.Config;
-import org.apache.cassandra.config.DatabaseDescriptor;
-import org.apache.cassandra.db.marshal.AbstractType;
-import org.apache.cassandra.db.marshal.BytesType;
-import org.apache.cassandra.hadoop.cql3.CqlConfigHelper;
-import org.apache.cassandra.io.sstable.SSTableLoader;
-import org.apache.cassandra.io.sstable.SSTableSimpleUnsortedWriter;
-import org.apache.cassandra.streaming.StreamState;
-import org.apache.cassandra.thrift.Column;
-import org.apache.cassandra.thrift.CounterColumn;
-import org.apache.cassandra.thrift.Mutation;
-import org.apache.cassandra.utils.NativeSSTableLoaderClient;
-import org.apache.cassandra.utils.OutputHandler;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.mapreduce.RecordWriter;
-import org.apache.hadoop.mapreduce.TaskAttemptContext;
-import org.apache.hadoop.util.Progressable;
-
-@Deprecated
-public final class BulkRecordWriter extends RecordWriter<ByteBuffer, List<Mutation>>
- implements org.apache.hadoop.mapred.RecordWriter<ByteBuffer, List<Mutation>>
-{
- public final static String OUTPUT_LOCATION = "mapreduce.output.bulkoutputformat.localdir";
- public final static String BUFFER_SIZE_IN_MB = "mapreduce.output.bulkoutputformat.buffersize";
- public final static String STREAM_THROTTLE_MBITS = "mapreduce.output.bulkoutputformat.streamthrottlembits";
- public final static String MAX_FAILED_HOSTS = "mapreduce.output.bulkoutputformat.maxfailedhosts";
-
- private final Logger logger = LoggerFactory.getLogger(BulkRecordWriter.class);
-
- protected final Configuration conf;
- protected final int maxFailures;
- protected final int bufferSize;
- protected Closeable writer;
- protected SSTableLoader loader;
- protected Progressable progress;
- protected TaskAttemptContext context;
- private File outputDir;
-
-
- private enum CFType
- {
- NORMAL,
- SUPER,
- }
-
- private enum ColType
- {
- NORMAL,
- COUNTER
- }
-
- private CFType cfType;
- private ColType colType;
-
- BulkRecordWriter(TaskAttemptContext context)
- {
-
- this(HadoopCompat.getConfiguration(context));
- this.context = context;
- }
-
- BulkRecordWriter(Configuration conf, Progressable progress)
- {
- this(conf);
- this.progress = progress;
- }
-
- BulkRecordWriter(Configuration conf)
- {
- Config.setOutboundBindAny(true);
- this.conf = conf;
- DatabaseDescriptor.setStreamThroughputOutboundMegabitsPerSec(Integer.parseInt(conf.get(STREAM_THROTTLE_MBITS, "0")));
- maxFailures = Integer.parseInt(conf.get(MAX_FAILED_HOSTS, "0"));
- bufferSize = Integer.parseInt(conf.get(BUFFER_SIZE_IN_MB, "64"));
- }
-
- protected String getOutputLocation() throws IOException
- {
- String dir = conf.get(OUTPUT_LOCATION, System.getProperty("java.io.tmpdir"));
- if (dir == null)
- throw new IOException("Output directory not defined, if hadoop is not setting java.io.tmpdir then define " + OUTPUT_LOCATION);
- return dir;
- }
-
- private void setTypes(Mutation mutation)
- {
- if (cfType == null)
- {
- if (mutation.getColumn_or_supercolumn().isSetSuper_column() || mutation.getColumn_or_supercolumn().isSetCounter_super_column())
- cfType = CFType.SUPER;
- else
- cfType = CFType.NORMAL;
- if (mutation.getColumn_or_supercolumn().isSetCounter_column() || mutation.getColumn_or_supercolumn().isSetCounter_super_column())
- colType = ColType.COUNTER;
- else
- colType = ColType.NORMAL;
- }
- }
-
- private void prepareWriter() throws IOException
- {
- if (outputDir == null)
- {
- String keyspace = ConfigHelper.getOutputKeyspace(conf);
- //dir must be named by ks/cf for the loader
- outputDir = new File(getOutputLocation() + File.separator + keyspace + File.separator + ConfigHelper.getOutputColumnFamily(conf));
- outputDir.mkdirs();
- }
-
- if (writer == null)
- {
- AbstractType<?> subcomparator = null;
-
- if (cfType == CFType.SUPER)
- subcomparator = BytesType.instance;
-
- writer = new SSTableSimpleUnsortedWriter(
- outputDir,
- ConfigHelper.getOutputPartitioner(conf),
- ConfigHelper.getOutputKeyspace(conf),
- ConfigHelper.getOutputColumnFamily(conf),
- BytesType.instance,
- subcomparator,
- Integer.parseInt(conf.get(BUFFER_SIZE_IN_MB, "64")),
- ConfigHelper.getOutputCompressionParamaters(conf));
-
- this.loader = new SSTableLoader(outputDir, new ExternalClient(conf), new NullOutputHandler());
- }
- }
-
- @Override
- public void close(TaskAttemptContext context) throws IOException
- {
- close();
- }
-
- /** Fills the deprecated RecordWriter interface for streaming. */
- @Deprecated
- public void close(org.apache.hadoop.mapred.Reporter reporter) throws IOException
- {
- close();
- }
-
- private void close() throws IOException
- {
- if (writer != null)
- {
- writer.close();
- Future<StreamState> future = loader.stream();
- while (true)
- {
- try
- {
- future.get(1000, TimeUnit.MILLISECONDS);
- break;
- }
- catch (ExecutionException | TimeoutException te)
- {
- if (null != progress)
- progress.progress();
- if (null != context)
- HadoopCompat.progress(context);
- }
- catch (InterruptedException e)
- {
- throw new IOException(e);
- }
- }
- if (loader.getFailedHosts().size() > 0)
- {
- if (loader.getFailedHosts().size() > maxFailures)
- throw new IOException("Too many hosts failed: " + loader.getFailedHosts());
- else
- logger.warn("Some hosts failed: {}", loader.getFailedHosts());
- }
- }
- }
-
- @Override
- public void write(ByteBuffer keybuff, List<Mutation> value) throws IOException
- {
- setTypes(value.get(0));
- prepareWriter();
- SSTableSimpleUnsortedWriter ssWriter = (SSTableSimpleUnsortedWriter) writer;
- ssWriter.newRow(keybuff);
- for (Mutation mut : value)
- {
- if (cfType == CFType.SUPER)
- {
- ssWriter.newSuperColumn(mut.getColumn_or_supercolumn().getSuper_column().name);
- if (colType == ColType.COUNTER)
- for (CounterColumn column : mut.getColumn_or_supercolumn().getCounter_super_column().columns)
- ssWriter.addCounterColumn(column.name, column.value);
- else
- {
- for (Column column : mut.getColumn_or_supercolumn().getSuper_column().columns)
- {
- if(column.ttl == 0)
- ssWriter.addColumn(column.name, column.value, column.timestamp);
- else
- ssWriter.addExpiringColumn(column.name, column.value, column.timestamp, column.ttl, System.currentTimeMillis() + ((long)column.ttl * 1000));
- }
- }
- }
- else
- {
- if (colType == ColType.COUNTER)
- ssWriter.addCounterColumn(mut.getColumn_or_supercolumn().counter_column.name, mut.getColumn_or_supercolumn().counter_column.value);
- else
- {
- if(mut.getColumn_or_supercolumn().column.ttl == 0)
- ssWriter.addColumn(mut.getColumn_or_supercolumn().column.name, mut.getColumn_or_supercolumn().column.value, mut.getColumn_or_supercolumn().column.timestamp);
- else
- ssWriter.addExpiringColumn(mut.getColumn_or_supercolumn().column.name, mut.getColumn_or_supercolumn().column.value, mut.getColumn_or_supercolumn().column.timestamp, mut.getColumn_or_supercolumn().column.ttl, System.currentTimeMillis() + ((long)(mut.getColumn_or_supercolumn().column.ttl) * 1000));
- }
- }
- if (null != progress)
- progress.progress();
- if (null != context)
- HadoopCompat.progress(context);
- }
- }
-
- public static class ExternalClient extends NativeSSTableLoaderClient
- {
- public ExternalClient(Configuration conf)
- {
- super(resolveHostAddresses(conf),
- CqlConfigHelper.getOutputNativePort(conf),
- ConfigHelper.getOutputKeyspaceUserName(conf),
- ConfigHelper.getOutputKeyspacePassword(conf),
- CqlConfigHelper.getSSLOptions(conf).orNull());
- }
-
- private static Collection<InetAddress> resolveHostAddresses(Configuration conf)
- {
- Set<InetAddress> addresses = new HashSet<>();
-
- for (String host : ConfigHelper.getOutputInitialAddress(conf).split(","))
- {
- try
- {
- addresses.add(InetAddress.getByName(host));
- }
- catch (UnknownHostException e)
- {
- throw new RuntimeException(e);
- }
- }
-
- return addresses;
- }
- }
-
- public static class NullOutputHandler implements OutputHandler
- {
- public void output(String msg) {}
- public void debug(String msg) {}
- public void warn(String msg) {}
- public void warn(String msg, Throwable th) {}
- }
-}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/446e2537/src/java/org/apache/cassandra/hadoop/ColumnFamilyInputFormat.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/hadoop/ColumnFamilyInputFormat.java b/src/java/org/apache/cassandra/hadoop/ColumnFamilyInputFormat.java
deleted file mode 100644
index 4662fa5..0000000
--- a/src/java/org/apache/cassandra/hadoop/ColumnFamilyInputFormat.java
+++ /dev/null
@@ -1,125 +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.cassandra.hadoop;
-
-import java.io.IOException;
-import java.nio.ByteBuffer;
-import java.util.*;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import org.apache.cassandra.auth.PasswordAuthenticator;
-import org.apache.cassandra.thrift.*;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.mapred.JobConf;
-import org.apache.hadoop.mapred.Reporter;
-import org.apache.hadoop.mapreduce.InputSplit;
-import org.apache.hadoop.mapreduce.RecordReader;
-import org.apache.hadoop.mapreduce.TaskAttemptContext;
-import org.apache.hadoop.mapreduce.TaskAttemptID;
-import org.apache.thrift.protocol.TBinaryProtocol;
-import org.apache.thrift.protocol.TProtocol;
-import org.apache.thrift.transport.TTransport;
-import org.apache.thrift.transport.TTransportException;
-
-/**
- * Hadoop InputFormat allowing map/reduce against Cassandra rows within one ColumnFamily.
- *
- * At minimum, you need to set the CF and predicate (description of columns to extract from each row)
- * in your Hadoop job Configuration. The ConfigHelper class is provided to make this
- * simple:
- * ConfigHelper.setInputColumnFamily
- * ConfigHelper.setInputSlicePredicate
- *
- * You can also configure the number of rows per InputSplit with
- * ConfigHelper.setInputSplitSize
- * This should be "as big as possible, but no bigger." Each InputSplit is read from Cassandra
- * with multiple get_slice_range queries, and the per-call overhead of get_slice_range is high,
- * so larger split sizes are better -- but if it is too large, you will run out of memory.
- *
- * The default split size is 64k rows.
- */
-@Deprecated
-public class ColumnFamilyInputFormat extends AbstractColumnFamilyInputFormat<ByteBuffer, SortedMap<ByteBuffer, ColumnFamilyRecordReader.Column>>
-{
- private static final Logger logger = LoggerFactory.getLogger(ColumnFamilyInputFormat.class);
-
- @SuppressWarnings("resource")
- public static Cassandra.Client createAuthenticatedClient(String location, int port, Configuration conf) throws Exception
- {
- logger.debug("Creating authenticated client for CF input format");
- TTransport transport;
- try
- {
- transport = ConfigHelper.getClientTransportFactory(conf).openTransport(location, port);
- }
- catch (Exception e)
- {
- throw new TTransportException("Failed to open a transport to " + location + ":" + port + ".", e);
- }
- TProtocol binaryProtocol = new TBinaryProtocol(transport, true, true);
- Cassandra.Client client = new Cassandra.Client(binaryProtocol);
-
- // log in
- client.set_keyspace(ConfigHelper.getInputKeyspace(conf));
- if ((ConfigHelper.getInputKeyspaceUserName(conf) != null) && (ConfigHelper.getInputKeyspacePassword(conf) != null))
- {
- Map<String, String> creds = new HashMap<String, String>();
- creds.put(PasswordAuthenticator.USERNAME_KEY, ConfigHelper.getInputKeyspaceUserName(conf));
- creds.put(PasswordAuthenticator.PASSWORD_KEY, ConfigHelper.getInputKeyspacePassword(conf));
- AuthenticationRequest authRequest = new AuthenticationRequest(creds);
- client.login(authRequest);
- }
- logger.debug("Authenticated client for CF input format created successfully");
- return client;
- }
-
- public RecordReader<ByteBuffer, SortedMap<ByteBuffer, ColumnFamilyRecordReader.Column>> createRecordReader(InputSplit inputSplit, TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException
- {
- return new ColumnFamilyRecordReader();
- }
-
- public org.apache.hadoop.mapred.RecordReader<ByteBuffer, SortedMap<ByteBuffer, ColumnFamilyRecordReader.Column>> getRecordReader(org.apache.hadoop.mapred.InputSplit split, JobConf jobConf, final Reporter reporter) throws IOException
- {
- TaskAttemptContext tac = HadoopCompat.newMapContext(
- jobConf,
- TaskAttemptID.forName(jobConf.get(MAPRED_TASK_ID)),
- null,
- null,
- null,
- new ReporterWrapper(reporter),
- null);
-
- ColumnFamilyRecordReader recordReader = new ColumnFamilyRecordReader(jobConf.getInt(CASSANDRA_HADOOP_MAX_KEY_SIZE, CASSANDRA_HADOOP_MAX_KEY_SIZE_DEFAULT));
- recordReader.initialize((org.apache.hadoop.mapreduce.InputSplit)split, tac);
- return recordReader;
- }
-
- @Override
- protected void validateConfiguration(Configuration conf)
- {
- super.validateConfiguration(conf);
-
- if (ConfigHelper.getInputSlicePredicate(conf) == null)
- {
- throw new UnsupportedOperationException("you must set the predicate with setInputSlicePredicate");
- }
- }
-
-}
[2/4] cassandra git commit: Remove deprecated legacy Hadoop code
Posted by al...@apache.org.
http://git-wip-us.apache.org/repos/asf/cassandra/blob/446e2537/src/java/org/apache/cassandra/hadoop/pig/CassandraStorage.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/hadoop/pig/CassandraStorage.java b/src/java/org/apache/cassandra/hadoop/pig/CassandraStorage.java
deleted file mode 100644
index 7bf43ef..0000000
--- a/src/java/org/apache/cassandra/hadoop/pig/CassandraStorage.java
+++ /dev/null
@@ -1,1397 +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.cassandra.hadoop.pig;
-
-import java.io.IOException;
-import java.io.UnsupportedEncodingException;
-import java.net.URLDecoder;
-import java.nio.ByteBuffer;
-import java.nio.charset.CharacterCodingException;
-import java.util.*;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import org.apache.cassandra.auth.PasswordAuthenticator;
-import org.apache.cassandra.config.CFMetaData;
-import org.apache.cassandra.config.ColumnDefinition;
-import org.apache.cassandra.db.marshal.*;
-import org.apache.cassandra.db.SystemKeyspace;
-import org.apache.cassandra.exceptions.ConfigurationException;
-import org.apache.cassandra.exceptions.InvalidRequestException;
-import org.apache.cassandra.exceptions.SyntaxException;
-import org.apache.cassandra.hadoop.ColumnFamilyRecordReader;
-import org.apache.cassandra.hadoop.ConfigHelper;
-import org.apache.cassandra.hadoop.HadoopCompat;
-import org.apache.cassandra.schema.LegacySchemaTables;
-import org.apache.cassandra.serializers.CollectionSerializer;
-import org.apache.cassandra.thrift.*;
-import org.apache.cassandra.utils.ByteBufferUtil;
-import org.apache.cassandra.utils.FBUtilities;
-import org.apache.cassandra.utils.Hex;
-import org.apache.cassandra.utils.UUIDGen;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.mapreduce.*;
-import org.apache.pig.Expression;
-import org.apache.pig.LoadFunc;
-import org.apache.pig.LoadMetadata;
-import org.apache.pig.ResourceSchema;
-import org.apache.pig.ResourceStatistics;
-import org.apache.pig.StoreFuncInterface;
-import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigSplit;
-import org.apache.pig.data.*;
-import org.apache.pig.impl.util.UDFContext;
-import org.apache.pig.ResourceSchema.ResourceFieldSchema;
-import org.apache.thrift.TDeserializer;
-import org.apache.thrift.TException;
-import org.apache.thrift.TSerializer;
-import org.apache.thrift.protocol.TBinaryProtocol;
-
-/**
- * A LoadStoreFunc for retrieving data from and storing data to Cassandra
- *
- * A row from a standard CF will be returned as nested tuples: (key, ((name1, val1), (name2, val2))).
- */
-@Deprecated
-public class CassandraStorage extends LoadFunc implements StoreFuncInterface, LoadMetadata
-{
- public final static String PIG_ALLOW_DELETES = "PIG_ALLOW_DELETES";
- public final static String PIG_WIDEROW_INPUT = "PIG_WIDEROW_INPUT";
- public final static String PIG_USE_SECONDARY = "PIG_USE_SECONDARY";
-
- private final static ByteBuffer BOUND = ByteBufferUtil.EMPTY_BYTE_BUFFER;
- private static final Logger logger = LoggerFactory.getLogger(CassandraStorage.class);
-
- private ByteBuffer slice_start = BOUND;
- private ByteBuffer slice_end = BOUND;
- private boolean slice_reverse = false;
- private boolean allow_deletes = false;
-
- private RecordReader<ByteBuffer, Map<ByteBuffer, ColumnFamilyRecordReader.Column>> reader;
- private RecordWriter<ByteBuffer, List<Mutation>> writer;
-
- private boolean widerows = false;
- private int limit;
-
- protected String DEFAULT_INPUT_FORMAT;
- protected String DEFAULT_OUTPUT_FORMAT;
-
- protected enum MarshallerType { COMPARATOR, DEFAULT_VALIDATOR, KEY_VALIDATOR, SUBCOMPARATOR };
-
- protected String username;
- protected String password;
- protected String keyspace;
- protected String column_family;
- protected String loadSignature;
- protected String storeSignature;
-
- protected Configuration conf;
- protected String inputFormatClass;
- protected String outputFormatClass;
- protected int splitSize = 64 * 1024;
- protected String partitionerClass;
- protected boolean usePartitionFilter = false;
- protected String initHostAddress;
- protected String rpcPort;
- protected int nativeProtocolVersion = 1;
-
- // wide row hacks
- private ByteBuffer lastKey;
- private Map<ByteBuffer, ColumnFamilyRecordReader.Column> lastRow;
-
- public CassandraStorage()
- {
- this(1024);
- }
-
- /**@param limit number of columns to fetch in a slice */
- public CassandraStorage(int limit)
- {
- super();
- this.limit = limit;
- DEFAULT_INPUT_FORMAT = "org.apache.cassandra.hadoop.ColumnFamilyInputFormat";
- DEFAULT_OUTPUT_FORMAT = "org.apache.cassandra.hadoop.ColumnFamilyOutputFormat";
- }
-
- public int getLimit()
- {
- return limit;
- }
-
- @Override
- public void prepareToRead(RecordReader reader, PigSplit split)
- {
- this.reader = reader;
- }
-
- /** read wide row*/
- public Tuple getNextWide() throws IOException
- {
- CfDef cfDef = getCfDef(loadSignature);
- ByteBuffer key = null;
- Tuple tuple = null;
- DefaultDataBag bag = new DefaultDataBag();
- try
- {
- while(true)
- {
- boolean hasNext = reader.nextKeyValue();
- if (!hasNext)
- {
- if (tuple == null)
- tuple = TupleFactory.getInstance().newTuple();
-
- if (lastRow != null)
- {
- if (tuple.size() == 0) // lastRow is a new one
- {
- key = reader.getCurrentKey();
- tuple = keyToTuple(key, cfDef, parseType(cfDef.getKey_validation_class()));
- }
- for (Map.Entry<ByteBuffer, ColumnFamilyRecordReader.Column> entry : lastRow.entrySet())
- {
- bag.add(columnToTuple(entry.getValue(), cfDef, parseType(cfDef.getComparator_type())));
- }
- lastKey = null;
- lastRow = null;
- tuple.append(bag);
- return tuple;
- }
- else
- {
- if (tuple.size() == 1) // rare case of just one wide row, key already set
- {
- tuple.append(bag);
- return tuple;
- }
- else
- return null;
- }
- }
- if (key != null && !(reader.getCurrentKey()).equals(key)) // key changed
- {
- // read too much, hold on to it for next time
- lastKey = reader.getCurrentKey();
- lastRow = reader.getCurrentValue();
- // but return what we have so far
- tuple.append(bag);
- return tuple;
- }
- if (key == null) // only set the key on the first iteration
- {
- key = reader.getCurrentKey();
- if (lastKey != null && !(key.equals(lastKey))) // last key only had one value
- {
- if (tuple == null)
- tuple = keyToTuple(lastKey, cfDef, parseType(cfDef.getKey_validation_class()));
- else
- addKeyToTuple(tuple, lastKey, cfDef, parseType(cfDef.getKey_validation_class()));
- for (Map.Entry<ByteBuffer, ColumnFamilyRecordReader.Column> entry : lastRow.entrySet())
- {
- bag.add(columnToTuple(entry.getValue(), cfDef, parseType(cfDef.getComparator_type())));
- }
- tuple.append(bag);
- lastKey = key;
- lastRow = reader.getCurrentValue();
- return tuple;
- }
- if (tuple == null)
- tuple = keyToTuple(key, cfDef, parseType(cfDef.getKey_validation_class()));
- else
- addKeyToTuple(tuple, lastKey, cfDef, parseType(cfDef.getKey_validation_class()));
- }
- SortedMap<ByteBuffer, ColumnFamilyRecordReader.Column> row =
- (SortedMap<ByteBuffer, ColumnFamilyRecordReader.Column>)reader.getCurrentValue();
- if (lastRow != null) // prepend what was read last time
- {
- for (Map.Entry<ByteBuffer, ColumnFamilyRecordReader.Column> entry : lastRow.entrySet())
- {
- bag.add(columnToTuple(entry.getValue(), cfDef, parseType(cfDef.getComparator_type())));
- }
- lastKey = null;
- lastRow = null;
- }
- for (Map.Entry<ByteBuffer, ColumnFamilyRecordReader.Column> entry : row.entrySet())
- {
- bag.add(columnToTuple(entry.getValue(), cfDef, parseType(cfDef.getComparator_type())));
- }
- }
- }
- catch (InterruptedException e)
- {
- throw new IOException(e.getMessage());
- }
- }
-
- /** read next row */
- public Tuple getNext() throws IOException
- {
- if (widerows)
- return getNextWide();
- try
- {
- // load the next pair
- if (!reader.nextKeyValue())
- return null;
-
- CfDef cfDef = getCfDef(loadSignature);
- ByteBuffer key = reader.getCurrentKey();
- Map<ByteBuffer, ColumnFamilyRecordReader.Column> cf = reader.getCurrentValue();
- assert key != null && cf != null;
-
- // output tuple, will hold the key, each indexed column in a tuple, then a bag of the rest
- // NOTE: we're setting the tuple size here only for the key so we can use setTupleValue on it
-
- Tuple tuple = keyToTuple(key, cfDef, parseType(cfDef.getKey_validation_class()));
- DefaultDataBag bag = new DefaultDataBag();
- // we must add all the indexed columns first to match the schema
- Map<ByteBuffer, Boolean> added = new HashMap<ByteBuffer, Boolean>(cfDef.column_metadata.size());
- // take care to iterate these in the same order as the schema does
- for (ColumnDef cdef : cfDef.column_metadata)
- {
- boolean hasColumn = false;
- boolean cql3Table = false;
- try
- {
- hasColumn = cf.containsKey(cdef.name);
- }
- catch (Exception e)
- {
- cql3Table = true;
- }
- if (hasColumn)
- {
- tuple.append(columnToTuple(cf.get(cdef.name), cfDef, parseType(cfDef.getComparator_type())));
- }
- else if (!cql3Table)
- { // otherwise, we need to add an empty tuple to take its place
- tuple.append(TupleFactory.getInstance().newTuple());
- }
- added.put(cdef.name, true);
- }
- // now add all the other columns
- for (Map.Entry<ByteBuffer, ColumnFamilyRecordReader.Column> entry : cf.entrySet())
- {
- if (!added.containsKey(entry.getKey()))
- bag.add(columnToTuple(entry.getValue(), cfDef, parseType(cfDef.getComparator_type())));
- }
- tuple.append(bag);
- // finally, special top-level indexes if needed
- if (usePartitionFilter)
- {
- for (ColumnDef cdef : getIndexes())
- {
- Tuple throwaway = columnToTuple(cf.get(cdef.name), cfDef, parseType(cfDef.getComparator_type()));
- tuple.append(throwaway.get(1));
- }
- }
- return tuple;
- }
- catch (InterruptedException e)
- {
- throw new IOException(e.getMessage());
- }
- }
-
- /** write next row */
- public void putNext(Tuple t) throws IOException
- {
- /*
- We support two cases for output:
- First, the original output:
- (key, (name, value), (name,value), {(name,value)}) (tuples or bag is optional)
- For supers, we only accept the original output.
- */
-
- if (t.size() < 1)
- {
- // simply nothing here, we can't even delete without a key
- logger.warn("Empty output skipped, filter empty tuples to suppress this warning");
- return;
- }
- ByteBuffer key = objToBB(t.get(0));
- if (t.getType(1) == DataType.TUPLE)
- writeColumnsFromTuple(key, t, 1);
- else if (t.getType(1) == DataType.BAG)
- {
- if (t.size() > 2)
- throw new IOException("No arguments allowed after bag");
- writeColumnsFromBag(key, (DataBag) t.get(1));
- }
- else
- throw new IOException("Second argument in output must be a tuple or bag");
- }
-
- /** set hadoop cassandra connection settings */
- protected void setConnectionInformation() throws IOException
- {
- StorageHelper.setConnectionInformation(conf);
- if (System.getenv(StorageHelper.PIG_INPUT_FORMAT) != null)
- inputFormatClass = getFullyQualifiedClassName(System.getenv(StorageHelper.PIG_INPUT_FORMAT));
- else
- inputFormatClass = DEFAULT_INPUT_FORMAT;
- if (System.getenv(StorageHelper.PIG_OUTPUT_FORMAT) != null)
- outputFormatClass = getFullyQualifiedClassName(System.getenv(StorageHelper.PIG_OUTPUT_FORMAT));
- else
- outputFormatClass = DEFAULT_OUTPUT_FORMAT;
- if (System.getenv(PIG_ALLOW_DELETES) != null)
- allow_deletes = Boolean.parseBoolean(System.getenv(PIG_ALLOW_DELETES));
- }
-
- /** get the full class name */
- protected String getFullyQualifiedClassName(String classname)
- {
- return classname.contains(".") ? classname : "org.apache.cassandra.hadoop." + classname;
- }
-
- /** set read configuration settings */
- public void setLocation(String location, Job job) throws IOException
- {
- conf = HadoopCompat.getConfiguration(job);
- setLocationFromUri(location);
-
- if (ConfigHelper.getInputSlicePredicate(conf) == null)
- {
- SliceRange range = new SliceRange(slice_start, slice_end, slice_reverse, limit);
- SlicePredicate predicate = new SlicePredicate().setSlice_range(range);
- ConfigHelper.setInputSlicePredicate(conf, predicate);
- }
- if (System.getenv(PIG_WIDEROW_INPUT) != null)
- widerows = Boolean.parseBoolean(System.getenv(PIG_WIDEROW_INPUT));
- if (System.getenv(PIG_USE_SECONDARY) != null)
- usePartitionFilter = Boolean.parseBoolean(System.getenv(PIG_USE_SECONDARY));
- if (System.getenv(StorageHelper.PIG_INPUT_SPLIT_SIZE) != null)
- {
- try
- {
- ConfigHelper.setInputSplitSize(conf, Integer.parseInt(System.getenv(StorageHelper.PIG_INPUT_SPLIT_SIZE)));
- }
- catch (NumberFormatException e)
- {
- throw new IOException("PIG_INPUT_SPLIT_SIZE is not a number", e);
- }
- }
-
- if (usePartitionFilter && getIndexExpressions() != null)
- ConfigHelper.setInputRange(conf, getIndexExpressions());
-
- if (username != null && password != null)
- ConfigHelper.setInputKeyspaceUserNameAndPassword(conf, username, password);
-
- if (splitSize > 0)
- ConfigHelper.setInputSplitSize(conf, splitSize);
- if (partitionerClass!= null)
- ConfigHelper.setInputPartitioner(conf, partitionerClass);
- if (rpcPort != null)
- ConfigHelper.setInputRpcPort(conf, rpcPort);
- if (initHostAddress != null)
- ConfigHelper.setInputInitialAddress(conf, initHostAddress);
-
- ConfigHelper.setInputColumnFamily(conf, keyspace, column_family, widerows);
- setConnectionInformation();
-
- if (ConfigHelper.getInputRpcPort(conf) == 0)
- throw new IOException("PIG_INPUT_RPC_PORT or PIG_RPC_PORT environment variable not set");
- if (ConfigHelper.getInputInitialAddress(conf) == null)
- throw new IOException("PIG_INPUT_INITIAL_ADDRESS or PIG_INITIAL_ADDRESS environment variable not set");
- if (ConfigHelper.getInputPartitioner(conf) == null)
- throw new IOException("PIG_INPUT_PARTITIONER or PIG_PARTITIONER environment variable not set");
- if (loadSignature == null)
- loadSignature = location;
- initSchema(loadSignature);
- }
-
- /** set store configuration settings */
- public void setStoreLocation(String location, Job job) throws IOException
- {
- conf = HadoopCompat.getConfiguration(job);
-
- // don't combine mappers to a single mapper per node
- conf.setBoolean("pig.noSplitCombination", true);
- setLocationFromUri(location);
-
- if (username != null && password != null)
- ConfigHelper.setOutputKeyspaceUserNameAndPassword(conf, username, password);
- if (splitSize > 0)
- ConfigHelper.setInputSplitSize(conf, splitSize);
- if (partitionerClass!= null)
- ConfigHelper.setOutputPartitioner(conf, partitionerClass);
- if (rpcPort != null)
- {
- ConfigHelper.setOutputRpcPort(conf, rpcPort);
- ConfigHelper.setInputRpcPort(conf, rpcPort);
- }
- if (initHostAddress != null)
- {
- ConfigHelper.setOutputInitialAddress(conf, initHostAddress);
- ConfigHelper.setInputInitialAddress(conf, initHostAddress);
- }
-
- ConfigHelper.setOutputColumnFamily(conf, keyspace, column_family);
- setConnectionInformation();
-
- if (ConfigHelper.getOutputRpcPort(conf) == 0)
- throw new IOException("PIG_OUTPUT_RPC_PORT or PIG_RPC_PORT environment variable not set");
- if (ConfigHelper.getOutputInitialAddress(conf) == null)
- throw new IOException("PIG_OUTPUT_INITIAL_ADDRESS or PIG_INITIAL_ADDRESS environment variable not set");
- if (ConfigHelper.getOutputPartitioner(conf) == null)
- throw new IOException("PIG_OUTPUT_PARTITIONER or PIG_PARTITIONER environment variable not set");
-
- // we have to do this again here for the check in writeColumnsFromTuple
- if (System.getenv(PIG_USE_SECONDARY) != null)
- usePartitionFilter = Boolean.parseBoolean(System.getenv(PIG_USE_SECONDARY));
-
- initSchema(storeSignature);
- }
-
- /** Methods to get the column family schema from Cassandra */
- protected void initSchema(String signature) throws IOException
- {
- Properties properties = UDFContext.getUDFContext().getUDFProperties(CassandraStorage.class);
-
- // Only get the schema if we haven't already gotten it
- if (!properties.containsKey(signature))
- {
- try
- {
- Cassandra.Client client = ConfigHelper.getClientFromInputAddressList(conf);
- client.set_keyspace(keyspace);
-
- if (username != null && password != null)
- {
- Map<String, String> credentials = new HashMap<String, String>(2);
- credentials.put(PasswordAuthenticator.USERNAME_KEY, username);
- credentials.put(PasswordAuthenticator.PASSWORD_KEY, password);
-
- try
- {
- client.login(new AuthenticationRequest(credentials));
- }
- catch (AuthenticationException e)
- {
- logger.error("Authentication exception: invalid username and/or password");
- throw new IOException(e);
- }
- }
-
- // compose the CfDef for the columfamily
- CfDef cfDef = getCfDef(client);
-
- if (cfDef != null)
- {
- StringBuilder sb = new StringBuilder();
- sb.append(cfdefToString(cfDef));
- properties.setProperty(signature, sb.toString());
- }
- else
- throw new IOException(String.format("Table '%s' not found in keyspace '%s'",
- column_family,
- keyspace));
- }
- catch (Exception e)
- {
- throw new IOException(e);
- }
- }
- }
-
- public void checkSchema(ResourceSchema schema) throws IOException
- {
- // we don't care about types, they all get casted to ByteBuffers
- }
-
- /** define the schema */
- public ResourceSchema getSchema(String location, Job job) throws IOException
- {
- setLocation(location, job);
- CfDef cfDef = getCfDef(loadSignature);
- if (cfDef.column_type.equals("Super"))
- return null;
- /*
- Our returned schema should look like this:
- (key, index1:(name, value), index2:(name, value), columns:{(name, value)})
- Which is to say, columns that have metadata will be returned as named tuples, but unknown columns will go into a bag.
- This way, wide rows can still be handled by the bag, but known columns can easily be referenced.
- */
-
- // top-level schema, no type
- ResourceSchema schema = new ResourceSchema();
-
- // get default marshallers and validators
- Map<MarshallerType, AbstractType> marshallers = getDefaultMarshallers(cfDef);
- Map<ByteBuffer,AbstractType> validators = getValidatorMap(cfDef);
-
- // add key
- ResourceFieldSchema keyFieldSchema = new ResourceFieldSchema();
- keyFieldSchema.setName("key");
- keyFieldSchema.setType(StorageHelper.getPigType(marshallers.get(MarshallerType.KEY_VALIDATOR)));
-
- ResourceSchema bagSchema = new ResourceSchema();
- ResourceFieldSchema bagField = new ResourceFieldSchema();
- bagField.setType(DataType.BAG);
- bagField.setName("columns");
- // inside the bag, place one tuple with the default comparator/validator schema
- ResourceSchema bagTupleSchema = new ResourceSchema();
- ResourceFieldSchema bagTupleField = new ResourceFieldSchema();
- bagTupleField.setType(DataType.TUPLE);
- ResourceFieldSchema bagcolSchema = new ResourceFieldSchema();
- ResourceFieldSchema bagvalSchema = new ResourceFieldSchema();
- bagcolSchema.setName("name");
- bagvalSchema.setName("value");
- bagcolSchema.setType(StorageHelper.getPigType(marshallers.get(MarshallerType.COMPARATOR)));
- bagvalSchema.setType(StorageHelper.getPigType(marshallers.get(MarshallerType.DEFAULT_VALIDATOR)));
- bagTupleSchema.setFields(new ResourceFieldSchema[] { bagcolSchema, bagvalSchema });
- bagTupleField.setSchema(bagTupleSchema);
- bagSchema.setFields(new ResourceFieldSchema[] { bagTupleField });
- bagField.setSchema(bagSchema);
-
- // will contain all fields for this schema
- List<ResourceFieldSchema> allSchemaFields = new ArrayList<ResourceFieldSchema>();
- // add the key first, then the indexed columns, and finally the bag
- allSchemaFields.add(keyFieldSchema);
-
- if (!widerows)
- {
- // defined validators/indexes
- for (ColumnDef cdef : cfDef.column_metadata)
- {
- // make a new tuple for each col/val pair
- ResourceSchema innerTupleSchema = new ResourceSchema();
- ResourceFieldSchema innerTupleField = new ResourceFieldSchema();
- innerTupleField.setType(DataType.TUPLE);
- innerTupleField.setSchema(innerTupleSchema);
- innerTupleField.setName(new String(cdef.getName()));
-
- ResourceFieldSchema idxColSchema = new ResourceFieldSchema();
- idxColSchema.setName("name");
- idxColSchema.setType(StorageHelper.getPigType(marshallers.get(MarshallerType.COMPARATOR)));
-
- ResourceFieldSchema valSchema = new ResourceFieldSchema();
- AbstractType validator = validators.get(cdef.name);
- if (validator == null)
- validator = marshallers.get(MarshallerType.DEFAULT_VALIDATOR);
- valSchema.setName("value");
- valSchema.setType(StorageHelper.getPigType(validator));
-
- innerTupleSchema.setFields(new ResourceFieldSchema[] { idxColSchema, valSchema });
- allSchemaFields.add(innerTupleField);
- }
- }
-
- // bag at the end for unknown columns
- allSchemaFields.add(bagField);
-
- // add top-level index elements if needed
- if (usePartitionFilter)
- {
- for (ColumnDef cdef : getIndexes())
- {
- ResourceFieldSchema idxSchema = new ResourceFieldSchema();
- idxSchema.setName("index_" + new String(cdef.getName()));
- AbstractType validator = validators.get(cdef.name);
- if (validator == null)
- validator = marshallers.get(MarshallerType.DEFAULT_VALIDATOR);
- idxSchema.setType(StorageHelper.getPigType(validator));
- allSchemaFields.add(idxSchema);
- }
- }
- // top level schema contains everything
- schema.setFields(allSchemaFields.toArray(new ResourceFieldSchema[allSchemaFields.size()]));
- return schema;
- }
-
- /** set partition filter */
- public void setPartitionFilter(Expression partitionFilter) throws IOException
- {
- UDFContext context = UDFContext.getUDFContext();
- Properties property = context.getUDFProperties(CassandraStorage.class);
- property.setProperty(StorageHelper.PARTITION_FILTER_SIGNATURE, indexExpressionsToString(filterToIndexExpressions(partitionFilter)));
- }
-
- /** prepare writer */
- public void prepareToWrite(RecordWriter writer)
- {
- this.writer = writer;
- }
-
- /** convert object to ByteBuffer */
- protected ByteBuffer objToBB(Object o)
- {
- if (o == null)
- return nullToBB();
- if (o instanceof java.lang.String)
- return ByteBuffer.wrap(new DataByteArray((String)o).get());
- if (o instanceof Integer)
- return Int32Type.instance.decompose((Integer)o);
- if (o instanceof Long)
- return LongType.instance.decompose((Long)o);
- if (o instanceof Float)
- return FloatType.instance.decompose((Float)o);
- if (o instanceof Double)
- return DoubleType.instance.decompose((Double)o);
- if (o instanceof UUID)
- return ByteBuffer.wrap(UUIDGen.decompose((UUID) o));
- if(o instanceof Tuple) {
- List<Object> objects = ((Tuple)o).getAll();
- //collections
- if (objects.size() > 0 && objects.get(0) instanceof String)
- {
- String collectionType = (String) objects.get(0);
- if ("set".equalsIgnoreCase(collectionType) ||
- "list".equalsIgnoreCase(collectionType))
- return objToListOrSetBB(objects.subList(1, objects.size()));
- else if ("map".equalsIgnoreCase(collectionType))
- return objToMapBB(objects.subList(1, objects.size()));
-
- }
- return objToCompositeBB(objects);
- }
-
- return ByteBuffer.wrap(((DataByteArray) o).get());
- }
-
- private ByteBuffer objToListOrSetBB(List<Object> objects)
- {
- List<ByteBuffer> serialized = new ArrayList<ByteBuffer>(objects.size());
- for(Object sub : objects)
- {
- ByteBuffer buffer = objToBB(sub);
- serialized.add(buffer);
- }
- // NOTE: using protocol v1 serialization format for collections so as to not break
- // compatibility. Not sure if that's the right thing.
- return CollectionSerializer.pack(serialized, objects.size(), 1);
- }
-
- private ByteBuffer objToMapBB(List<Object> objects)
- {
- List<ByteBuffer> serialized = new ArrayList<ByteBuffer>(objects.size() * 2);
- for(Object sub : objects)
- {
- List<Object> keyValue = ((Tuple)sub).getAll();
- for (Object entry: keyValue)
- {
- ByteBuffer buffer = objToBB(entry);
- serialized.add(buffer);
- }
- }
- // NOTE: using protocol v1 serialization format for collections so as to not break
- // compatibility. Not sure if that's the right thing.
- return CollectionSerializer.pack(serialized, objects.size(), 1);
- }
-
- private ByteBuffer objToCompositeBB(List<Object> objects)
- {
- List<ByteBuffer> serialized = new ArrayList<ByteBuffer>(objects.size());
- int totalLength = 0;
- for(Object sub : objects)
- {
- ByteBuffer buffer = objToBB(sub);
- serialized.add(buffer);
- totalLength += 2 + buffer.remaining() + 1;
- }
- ByteBuffer out = ByteBuffer.allocate(totalLength);
- for (ByteBuffer bb : serialized)
- {
- int length = bb.remaining();
- out.put((byte) ((length >> 8) & 0xFF));
- out.put((byte) (length & 0xFF));
- out.put(bb);
- out.put((byte) 0);
- }
- out.flip();
- return out;
- }
-
- /** write tuple data to cassandra */
- private void writeColumnsFromTuple(ByteBuffer key, Tuple t, int offset) throws IOException
- {
- ArrayList<Mutation> mutationList = new ArrayList<Mutation>();
- for (int i = offset; i < t.size(); i++)
- {
- if (t.getType(i) == DataType.BAG)
- writeColumnsFromBag(key, (DataBag) t.get(i));
- else if (t.getType(i) == DataType.TUPLE)
- {
- Tuple inner = (Tuple) t.get(i);
- if (inner.size() > 0) // may be empty, for an indexed column that wasn't present
- mutationList.add(mutationFromTuple(inner));
- }
- else if (!usePartitionFilter)
- {
- throw new IOException("Output type was not a bag or a tuple");
- }
- }
- if (mutationList.size() > 0)
- writeMutations(key, mutationList);
- }
-
- /** compose Cassandra mutation from tuple */
- private Mutation mutationFromTuple(Tuple t) throws IOException
- {
- Mutation mutation = new Mutation();
- if (t.get(1) == null)
- {
- if (allow_deletes)
- {
- mutation.deletion = new Deletion();
- mutation.deletion.predicate = new org.apache.cassandra.thrift.SlicePredicate();
- mutation.deletion.predicate.column_names = Arrays.asList(objToBB(t.get(0)));
- mutation.deletion.setTimestamp(FBUtilities.timestampMicros());
- }
- else
- throw new IOException("null found but deletes are disabled, set " + PIG_ALLOW_DELETES +
- "=true in environment or allow_deletes=true in URL to enable");
- }
- else
- {
- org.apache.cassandra.thrift.Column column = new org.apache.cassandra.thrift.Column();
- column.setName(objToBB(t.get(0)));
- column.setValue(objToBB(t.get(1)));
- column.setTimestamp(FBUtilities.timestampMicros());
- mutation.column_or_supercolumn = new ColumnOrSuperColumn();
- mutation.column_or_supercolumn.column = column;
- }
- return mutation;
- }
-
- /** write bag data to Cassandra */
- private void writeColumnsFromBag(ByteBuffer key, DataBag bag) throws IOException
- {
- List<Mutation> mutationList = new ArrayList<Mutation>();
- for (Tuple pair : bag)
- {
- Mutation mutation = new Mutation();
- if (DataType.findType(pair.get(1)) == DataType.BAG) // supercolumn
- {
- SuperColumn sc = new SuperColumn();
- sc.setName(objToBB(pair.get(0)));
- List<org.apache.cassandra.thrift.Column> columns = new ArrayList<org.apache.cassandra.thrift.Column>();
- for (Tuple subcol : (DataBag) pair.get(1))
- {
- org.apache.cassandra.thrift.Column column = new org.apache.cassandra.thrift.Column();
- column.setName(objToBB(subcol.get(0)));
- column.setValue(objToBB(subcol.get(1)));
- column.setTimestamp(FBUtilities.timestampMicros());
- columns.add(column);
- }
- if (columns.isEmpty())
- {
- if (allow_deletes)
- {
- mutation.deletion = new Deletion();
- mutation.deletion.super_column = objToBB(pair.get(0));
- mutation.deletion.setTimestamp(FBUtilities.timestampMicros());
- }
- else
- throw new IOException("SuperColumn deletion attempted with empty bag, but deletes are disabled, set " +
- PIG_ALLOW_DELETES + "=true in environment or allow_deletes=true in URL to enable");
- }
- else
- {
- sc.columns = columns;
- mutation.column_or_supercolumn = new ColumnOrSuperColumn();
- mutation.column_or_supercolumn.super_column = sc;
- }
- }
- else
- mutation = mutationFromTuple(pair);
- mutationList.add(mutation);
- // for wide rows, we need to limit the amount of mutations we write at once
- if (mutationList.size() >= 10) // arbitrary, CFOF will re-batch this up, and BOF won't care
- {
- writeMutations(key, mutationList);
- mutationList.clear();
- }
- }
- // write the last batch
- if (mutationList.size() > 0)
- writeMutations(key, mutationList);
- }
-
- /** write mutation to Cassandra */
- private void writeMutations(ByteBuffer key, List<Mutation> mutations) throws IOException
- {
- try
- {
- writer.write(key, mutations);
- }
- catch (InterruptedException e)
- {
- throw new IOException(e);
- }
- }
-
- /** get a list of columns with defined index*/
- protected List<ColumnDef> getIndexes() throws IOException
- {
- CfDef cfdef = getCfDef(loadSignature);
- List<ColumnDef> indexes = new ArrayList<ColumnDef>();
- for (ColumnDef cdef : cfdef.column_metadata)
- {
- if (cdef.index_type != null)
- indexes.add(cdef);
- }
- return indexes;
- }
-
- /** get a list of Cassandra IndexExpression from Pig expression */
- private List<IndexExpression> filterToIndexExpressions(Expression expression) throws IOException
- {
- List<IndexExpression> indexExpressions = new ArrayList<IndexExpression>();
- Expression.BinaryExpression be = (Expression.BinaryExpression)expression;
- ByteBuffer name = ByteBuffer.wrap(be.getLhs().toString().getBytes());
- ByteBuffer value = ByteBuffer.wrap(be.getRhs().toString().getBytes());
- switch (expression.getOpType())
- {
- case OP_EQ:
- indexExpressions.add(new IndexExpression(name, IndexOperator.EQ, value));
- break;
- case OP_GE:
- indexExpressions.add(new IndexExpression(name, IndexOperator.GTE, value));
- break;
- case OP_GT:
- indexExpressions.add(new IndexExpression(name, IndexOperator.GT, value));
- break;
- case OP_LE:
- indexExpressions.add(new IndexExpression(name, IndexOperator.LTE, value));
- break;
- case OP_LT:
- indexExpressions.add(new IndexExpression(name, IndexOperator.LT, value));
- break;
- case OP_AND:
- indexExpressions.addAll(filterToIndexExpressions(be.getLhs()));
- indexExpressions.addAll(filterToIndexExpressions(be.getRhs()));
- break;
- default:
- throw new IOException("Unsupported expression type: " + expression.getOpType().name());
- }
- return indexExpressions;
- }
-
- /** convert a list of index expression to string */
- private static String indexExpressionsToString(List<IndexExpression> indexExpressions) throws IOException
- {
- assert indexExpressions != null;
- // oh, you thought cfdefToString was awful?
- IndexClause indexClause = new IndexClause();
- indexClause.setExpressions(indexExpressions);
- indexClause.setStart_key("".getBytes());
- TSerializer serializer = new TSerializer(new TBinaryProtocol.Factory());
- try
- {
- return Hex.bytesToHex(serializer.serialize(indexClause));
- }
- catch (TException e)
- {
- throw new IOException(e);
- }
- }
-
- /** convert string to a list of index expression */
- private static List<IndexExpression> indexExpressionsFromString(String ie) throws IOException
- {
- assert ie != null;
- TDeserializer deserializer = new TDeserializer(new TBinaryProtocol.Factory());
- IndexClause indexClause = new IndexClause();
- try
- {
- deserializer.deserialize(indexClause, Hex.hexToBytes(ie));
- }
- catch (TException e)
- {
- throw new IOException(e);
- }
- return indexClause.getExpressions();
- }
-
- public ResourceStatistics getStatistics(String location, Job job)
- {
- return null;
- }
-
- public void cleanupOnFailure(String failure, Job job)
- {
- }
-
- public void cleanupOnSuccess(String location, Job job) throws IOException {
- }
-
-
- /** StoreFunc methods */
- public void setStoreFuncUDFContextSignature(String signature)
- {
- this.storeSignature = signature;
- }
-
- public String relToAbsPathForStoreLocation(String location, Path curDir) throws IOException
- {
- return relativeToAbsolutePath(location, curDir);
- }
-
- /** output format */
- public OutputFormat getOutputFormat() throws IOException
- {
- try
- {
- return FBUtilities.construct(outputFormatClass, "outputformat");
- }
- catch (ConfigurationException e)
- {
- throw new IOException(e);
- }
- }
-
-
- @Override
- public InputFormat getInputFormat() throws IOException
- {
- try
- {
- return FBUtilities.construct(inputFormatClass, "inputformat");
- }
- catch (ConfigurationException e)
- {
- throw new IOException(e);
- }
- }
-
- /** get a list of index expression */
- private List<IndexExpression> getIndexExpressions() throws IOException
- {
- UDFContext context = UDFContext.getUDFContext();
- Properties property = context.getUDFProperties(CassandraStorage.class);
- if (property.getProperty(StorageHelper.PARTITION_FILTER_SIGNATURE) != null)
- return indexExpressionsFromString(property.getProperty(StorageHelper.PARTITION_FILTER_SIGNATURE));
- else
- return null;
- }
-
- /** get a list of column for the column family */
- protected List<ColumnDef> getColumnMetadata(Cassandra.Client client)
- throws TException, CharacterCodingException, InvalidRequestException, ConfigurationException
- {
- return getColumnMeta(client, true, true);
- }
-
-
- /** get column meta data */
- protected List<ColumnDef> getColumnMeta(Cassandra.Client client, boolean cassandraStorage, boolean includeCompactValueColumn)
- throws TException,
- CharacterCodingException,
- ConfigurationException
- {
- String query = String.format("SELECT column_name, validator, index_type, type " +
- "FROM %s.%s " +
- "WHERE keyspace_name = '%s' AND columnfamily_name = '%s'",
- SystemKeyspace.NAME,
- LegacySchemaTables.COLUMNS,
- keyspace,
- column_family);
-
- CqlResult result = client.execute_cql3_query(ByteBufferUtil.bytes(query), Compression.NONE, ConsistencyLevel.ONE);
-
- List<CqlRow> rows = result.rows;
- List<ColumnDef> columnDefs = new ArrayList<ColumnDef>();
- if (rows == null || rows.isEmpty())
- {
- // if CassandraStorage, just return the empty list
- if (cassandraStorage)
- return columnDefs;
-
- // otherwise for CqlNativeStorage, check metadata for classic thrift tables
- CFMetaData cfm = getCFMetaData(keyspace, column_family, client);
- for (ColumnDefinition def : cfm.regularAndStaticColumns())
- {
- ColumnDef cDef = new ColumnDef();
- String columnName = def.name.toString();
- String type = def.type.toString();
- logger.debug("name: {}, type: {} ", columnName, type);
- cDef.name = ByteBufferUtil.bytes(columnName);
- cDef.validation_class = type;
- columnDefs.add(cDef);
- }
- // we may not need to include the value column for compact tables as we
- // could have already processed it as schema_columnfamilies.value_alias
- if (columnDefs.size() == 0 && includeCompactValueColumn && cfm.compactValueColumn() != null)
- {
- ColumnDefinition def = cfm.compactValueColumn();
- if ("value".equals(def.name.toString()))
- {
- ColumnDef cDef = new ColumnDef();
- cDef.name = def.name.bytes;
- cDef.validation_class = def.type.toString();
- columnDefs.add(cDef);
- }
- }
- return columnDefs;
- }
-
- Iterator<CqlRow> iterator = rows.iterator();
- while (iterator.hasNext())
- {
- CqlRow row = iterator.next();
- ColumnDef cDef = new ColumnDef();
- String type = ByteBufferUtil.string(row.getColumns().get(3).value);
- if (!type.equals("regular"))
- continue;
- cDef.setName(ByteBufferUtil.clone(row.getColumns().get(0).value));
- cDef.validation_class = ByteBufferUtil.string(row.getColumns().get(1).value);
- ByteBuffer indexType = row.getColumns().get(2).value;
- if (indexType != null)
- cDef.index_type = getIndexType(ByteBufferUtil.string(indexType));
- columnDefs.add(cDef);
- }
- return columnDefs;
- }
-
-
- /** get CFMetaData of a column family */
- protected CFMetaData getCFMetaData(String ks, String cf, Cassandra.Client client)
- throws TException, ConfigurationException
- {
- KsDef ksDef = client.describe_keyspace(ks);
- for (CfDef cfDef : ksDef.cf_defs)
- {
- if (cfDef.name.equalsIgnoreCase(cf))
- return ThriftConversion.fromThrift(cfDef);
- }
- return null;
- }
-
- /** get index type from string */
- protected IndexType getIndexType(String type)
- {
- type = type.toLowerCase();
- if ("keys".equals(type))
- return IndexType.KEYS;
- else if("custom".equals(type))
- return IndexType.CUSTOM;
- else if("composites".equals(type))
- return IndexType.COMPOSITES;
- else
- return null;
- }
-
- /** return partition keys */
- public String[] getPartitionKeys(String location, Job job) throws IOException
- {
- if (!usePartitionFilter)
- return null;
- List<ColumnDef> indexes = getIndexes();
- String[] partitionKeys = new String[indexes.size()];
- for (int i = 0; i < indexes.size(); i++)
- {
- partitionKeys[i] = new String(indexes.get(i).getName());
- }
- return partitionKeys;
- }
-
- /** convert key to a tuple */
- private Tuple keyToTuple(ByteBuffer key, CfDef cfDef, AbstractType comparator) throws IOException
- {
- Tuple tuple = TupleFactory.getInstance().newTuple(1);
- addKeyToTuple(tuple, key, cfDef, comparator);
- return tuple;
- }
-
- /** add key to a tuple */
- private void addKeyToTuple(Tuple tuple, ByteBuffer key, CfDef cfDef, AbstractType comparator) throws IOException
- {
- if( comparator instanceof AbstractCompositeType )
- {
- StorageHelper.setTupleValue(tuple, 0, composeComposite((AbstractCompositeType) comparator, key));
- }
- else
- {
- StorageHelper.setTupleValue(tuple, 0, StorageHelper.cassandraToObj(getDefaultMarshallers(cfDef).get(MarshallerType.KEY_VALIDATOR), key, nativeProtocolVersion));
- }
-
- }
-
- /** Deconstructs a composite type to a Tuple. */
- protected Tuple composeComposite(AbstractCompositeType comparator, ByteBuffer name) throws IOException
- {
- List<AbstractCompositeType.CompositeComponent> result = comparator.deconstruct(name);
- Tuple t = TupleFactory.getInstance().newTuple(result.size());
- for (int i=0; i<result.size(); i++)
- StorageHelper.setTupleValue(t, i, StorageHelper.cassandraToObj(result.get(i).comparator, result.get(i).value, nativeProtocolVersion));
-
- return t;
- }
-
- /** cassandra://[username:password@]<keyspace>/<columnfamily>[?slice_start=<start>&slice_end=<end>
- * [&reversed=true][&limit=1][&allow_deletes=true][&widerows=true]
- * [&use_secondary=true][&comparator=<comparator>][&partitioner=<partitioner>]]*/
- private void setLocationFromUri(String location) throws IOException
- {
- try
- {
- if (!location.startsWith("cassandra://"))
- throw new Exception("Bad scheme." + location);
-
- String[] urlParts = location.split("\\?");
- if (urlParts.length > 1)
- {
- Map<String, String> urlQuery = getQueryMap(urlParts[1]);
- AbstractType comparator = BytesType.instance;
- if (urlQuery.containsKey("comparator"))
- comparator = TypeParser.parse(urlQuery.get("comparator"));
- if (urlQuery.containsKey("slice_start"))
- slice_start = comparator.fromString(urlQuery.get("slice_start"));
- if (urlQuery.containsKey("slice_end"))
- slice_end = comparator.fromString(urlQuery.get("slice_end"));
- if (urlQuery.containsKey("reversed"))
- slice_reverse = Boolean.parseBoolean(urlQuery.get("reversed"));
- if (urlQuery.containsKey("limit"))
- limit = Integer.parseInt(urlQuery.get("limit"));
- if (urlQuery.containsKey("allow_deletes"))
- allow_deletes = Boolean.parseBoolean(urlQuery.get("allow_deletes"));
- if (urlQuery.containsKey("widerows"))
- widerows = Boolean.parseBoolean(urlQuery.get("widerows"));
- if (urlQuery.containsKey("use_secondary"))
- usePartitionFilter = Boolean.parseBoolean(urlQuery.get("use_secondary"));
- if (urlQuery.containsKey("split_size"))
- splitSize = Integer.parseInt(urlQuery.get("split_size"));
- if (urlQuery.containsKey("partitioner"))
- partitionerClass = urlQuery.get("partitioner");
- if (urlQuery.containsKey("init_address"))
- initHostAddress = urlQuery.get("init_address");
- if (urlQuery.containsKey("rpc_port"))
- rpcPort = urlQuery.get("rpc_port");
- }
- String[] parts = urlParts[0].split("/+");
- String[] credentialsAndKeyspace = parts[1].split("@");
- if (credentialsAndKeyspace.length > 1)
- {
- String[] credentials = credentialsAndKeyspace[0].split(":");
- username = credentials[0];
- password = credentials[1];
- keyspace = credentialsAndKeyspace[1];
- }
- else
- {
- keyspace = parts[1];
- }
- column_family = parts[2];
- }
- catch (Exception e)
- {
- throw new IOException("Expected 'cassandra://[username:password@]<keyspace>/<table>" +
- "[?slice_start=<start>&slice_end=<end>[&reversed=true][&limit=1]" +
- "[&allow_deletes=true][&widerows=true][&use_secondary=true]" +
- "[&comparator=<comparator>][&split_size=<size>][&partitioner=<partitioner>]" +
- "[&init_address=<host>][&rpc_port=<port>]]': " + e.getMessage());
- }
- }
-
-
- /** decompose the query to store the parameters in a map */
- public static Map<String, String> getQueryMap(String query) throws UnsupportedEncodingException
- {
- String[] params = query.split("&");
- Map<String, String> map = new HashMap<String, String>(params.length);
- for (String param : params)
- {
- String[] keyValue = param.split("=");
- map.put(keyValue[0], URLDecoder.decode(keyValue[1], "UTF-8"));
- }
- return map;
- }
-
- public ByteBuffer nullToBB()
- {
- return null;
- }
-
- /** return the CfInfo for the column family */
- protected CfDef getCfDef(Cassandra.Client client)
- throws TException,
- ConfigurationException,
- IOException
- {
- // get CF meta data
- String query = String.format("SELECT type, comparator, subcomparator, default_validator, key_validator " +
- "FROM %s.%s " +
- "WHERE keyspace_name = '%s' AND columnfamily_name = '%s'",
- SystemKeyspace.NAME,
- LegacySchemaTables.COLUMNFAMILIES,
- keyspace,
- column_family);
-
- CqlResult result = client.execute_cql3_query(ByteBufferUtil.bytes(query), Compression.NONE, ConsistencyLevel.ONE);
-
- if (result == null || result.rows == null || result.rows.isEmpty())
- return null;
-
- Iterator<CqlRow> iteraRow = result.rows.iterator();
- CfDef cfDef = new CfDef();
- cfDef.keyspace = keyspace;
- cfDef.name = column_family;
- if (iteraRow.hasNext())
- {
- CqlRow cqlRow = iteraRow.next();
-
- cfDef.column_type = ByteBufferUtil.string(cqlRow.columns.get(0).value);
- cfDef.comparator_type = ByteBufferUtil.string(cqlRow.columns.get(1).value);
- ByteBuffer subComparator = cqlRow.columns.get(2).value;
- if (subComparator != null)
- cfDef.subcomparator_type = ByteBufferUtil.string(subComparator);
- cfDef.default_validation_class = ByteBufferUtil.string(cqlRow.columns.get(3).value);
- cfDef.key_validation_class = ByteBufferUtil.string(cqlRow.columns.get(4).value);
- }
- cfDef.column_metadata = getColumnMetadata(client);
- return cfDef;
- }
-
- /** get the columnfamily definition for the signature */
- protected CfDef getCfDef(String signature) throws IOException
- {
- UDFContext context = UDFContext.getUDFContext();
- Properties property = context.getUDFProperties(CassandraStorage.class);
- String prop = property.getProperty(signature);
- return cfdefFromString(prop);
- }
-
- /** convert string back to CfDef */
- protected static CfDef cfdefFromString(String st) throws IOException
- {
- assert st != null;
- TDeserializer deserializer = new TDeserializer(new TBinaryProtocol.Factory());
- CfDef cfDef = new CfDef();
- try
- {
- deserializer.deserialize(cfDef, Hex.hexToBytes(st));
- }
- catch (TException e)
- {
- throw new IOException(e);
- }
- return cfDef;
- }
-
- /** convert CfDef to string */
- protected static String cfdefToString(CfDef cfDef) throws IOException
- {
- assert cfDef != null;
- // this is so awful it's kind of cool!
- TSerializer serializer = new TSerializer(new TBinaryProtocol.Factory());
- try
- {
- return Hex.bytesToHex(serializer.serialize(cfDef));
- }
- catch (TException e)
- {
- throw new IOException(e);
- }
- }
-
- /** parse the string to a cassandra data type */
- protected AbstractType parseType(String type) throws IOException
- {
- try
- {
- // always treat counters like longs, specifically CCT.compose is not what we need
- if (type != null && type.equals("org.apache.cassandra.db.marshal.CounterColumnType"))
- return LongType.instance;
- return TypeParser.parse(type);
- }
- catch (ConfigurationException e)
- {
- throw new IOException(e);
- }
- catch (SyntaxException e)
- {
- throw new IOException(e);
- }
- }
-
- /** convert a column to a tuple */
- protected Tuple columnToTuple(ColumnFamilyRecordReader.Column column, CfDef cfDef, AbstractType comparator) throws IOException
- {
- Tuple pair = TupleFactory.getInstance().newTuple(2);
-
- // name
- if(comparator instanceof AbstractCompositeType)
- StorageHelper.setTupleValue(pair, 0, composeComposite((AbstractCompositeType) comparator, column.name));
- else
- StorageHelper.setTupleValue(pair, 0, StorageHelper.cassandraToObj(comparator, column.name, nativeProtocolVersion));
-
- // value
- Map<ByteBuffer,AbstractType> validators = getValidatorMap(cfDef);
- if (validators.get(column.name) == null)
- {
- Map<MarshallerType, AbstractType> marshallers = getDefaultMarshallers(cfDef);
- StorageHelper.setTupleValue(pair, 1, StorageHelper.cassandraToObj(marshallers.get(MarshallerType.DEFAULT_VALIDATOR), column.value, nativeProtocolVersion));
- }
- else
- StorageHelper.setTupleValue(pair, 1, StorageHelper.cassandraToObj(validators.get(column.name), column.value, nativeProtocolVersion));
- return pair;
- }
-
- /** construct a map to store the mashaller type to cassandra data type mapping */
- protected Map<MarshallerType, AbstractType> getDefaultMarshallers(CfDef cfDef) throws IOException
- {
- Map<MarshallerType, AbstractType> marshallers = new EnumMap<MarshallerType, AbstractType>(MarshallerType.class);
- AbstractType comparator;
- AbstractType subcomparator;
- AbstractType default_validator;
- AbstractType key_validator;
-
- comparator = parseType(cfDef.getComparator_type());
- subcomparator = parseType(cfDef.getSubcomparator_type());
- default_validator = parseType(cfDef.getDefault_validation_class());
- key_validator = parseType(cfDef.getKey_validation_class());
-
- marshallers.put(MarshallerType.COMPARATOR, comparator);
- marshallers.put(MarshallerType.DEFAULT_VALIDATOR, default_validator);
- marshallers.put(MarshallerType.KEY_VALIDATOR, key_validator);
- marshallers.put(MarshallerType.SUBCOMPARATOR, subcomparator);
- return marshallers;
- }
-
- /** get the validators */
- protected Map<ByteBuffer, AbstractType> getValidatorMap(CfDef cfDef) throws IOException
- {
- Map<ByteBuffer, AbstractType> validators = new HashMap<ByteBuffer, AbstractType>();
- for (ColumnDef cd : cfDef.getColumn_metadata())
- {
- if (cd.getValidation_class() != null && !cd.getValidation_class().isEmpty())
- {
- AbstractType validator = null;
- try
- {
- validator = TypeParser.parse(cd.getValidation_class());
- if (validator instanceof CounterColumnType)
- validator = LongType.instance;
- validators.put(cd.name, validator);
- }
- catch (ConfigurationException e)
- {
- throw new IOException(e);
- }
- catch (SyntaxException e)
- {
- throw new IOException(e);
- }
- }
- }
- return validators;
- }
-}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/446e2537/test/pig/org/apache/cassandra/pig/PigTestBase.java
----------------------------------------------------------------------
diff --git a/test/pig/org/apache/cassandra/pig/PigTestBase.java b/test/pig/org/apache/cassandra/pig/PigTestBase.java
index 8c27f6c..f556e66 100644
--- a/test/pig/org/apache/cassandra/pig/PigTestBase.java
+++ b/test/pig/org/apache/cassandra/pig/PigTestBase.java
@@ -20,6 +20,8 @@ package org.apache.cassandra.pig;
import java.io.IOException;
+import com.datastax.driver.core.Cluster;
+import com.datastax.driver.core.Session;
import org.apache.cassandra.SchemaLoader;
import org.apache.cassandra.config.Schema;
import org.apache.cassandra.db.marshal.AbstractType;
@@ -27,9 +29,6 @@ import org.apache.cassandra.db.marshal.TypeParser;
import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.exceptions.SyntaxException;
import org.apache.cassandra.service.EmbeddedCassandraService;
-import org.apache.cassandra.thrift.Cassandra;
-import org.apache.cassandra.thrift.Compression;
-import org.apache.cassandra.thrift.ConsistencyLevel;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.hadoop.conf.Configuration;
import org.apache.pig.ExecType;
@@ -37,13 +36,6 @@ import org.apache.pig.PigServer;
import org.apache.pig.backend.hadoop.datastorage.ConfigurationUtil;
import org.apache.pig.impl.PigContext;
import org.apache.pig.test.MiniCluster;
-import org.apache.thrift.TException;
-import org.apache.thrift.protocol.TBinaryProtocol;
-import org.apache.thrift.protocol.TProtocol;
-import org.apache.thrift.transport.TFramedTransport;
-import org.apache.thrift.transport.TSocket;
-import org.apache.thrift.transport.TTransport;
-import org.apache.thrift.transport.TTransportException;
import org.junit.After;
import org.junit.AfterClass;
import org.junit.Before;
@@ -80,13 +72,10 @@ public class PigTestBase extends SchemaLoader
pig.shutdown();
}
- protected static Cassandra.Client getClient() throws TTransportException
+ protected static Session getClient()
{
- TTransport tr = new TFramedTransport(new TSocket("localhost", 9170));
- TProtocol proto = new TBinaryProtocol(tr);
- Cassandra.Client client = new Cassandra.Client(proto);
- tr.open();
- return client;
+ Cluster cluster = Cluster.builder().addContactPoints("localhost").withPort(9042).build();
+ return cluster.connect();
}
protected static void startCassandra() throws IOException
@@ -114,14 +103,14 @@ public class PigTestBase extends SchemaLoader
}
}
- protected static void executeCQLStatements(String[] statements) throws TException
+ protected static void executeCQLStatements(String[] statements)
{
- Cassandra.Client client = getClient();
+ Session client = getClient();
for (String statement : statements)
{
System.out.println("Executing statement: " + statement);
- client.execute_cql3_query(ByteBufferUtil.bytes(statement), Compression.NONE, ConsistencyLevel.ONE);
+ client.execute(statement);
}
}
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/446e2537/test/pig/org/apache/cassandra/pig/ThriftColumnFamilyDataTypeTest.java
----------------------------------------------------------------------
diff --git a/test/pig/org/apache/cassandra/pig/ThriftColumnFamilyDataTypeTest.java b/test/pig/org/apache/cassandra/pig/ThriftColumnFamilyDataTypeTest.java
index 3ddb94e..273cdff 100644
--- a/test/pig/org/apache/cassandra/pig/ThriftColumnFamilyDataTypeTest.java
+++ b/test/pig/org/apache/cassandra/pig/ThriftColumnFamilyDataTypeTest.java
@@ -24,10 +24,8 @@ import org.apache.cassandra.db.marshal.TimeUUIDType;
import org.apache.cassandra.db.marshal.UUIDType;
import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.utils.Hex;
-import org.apache.pig.data.DataBag;
import org.apache.pig.data.DataByteArray;
import org.apache.pig.data.Tuple;
-import org.apache.thrift.TException;
import org.junit.BeforeClass;
import org.junit.Test;
@@ -69,7 +67,7 @@ public class ThriftColumnFamilyDataTypeTest extends PigTestBase
};
@BeforeClass
- public static void setup() throws IOException, ConfigurationException, TException
+ public static void setup() throws IOException, ConfigurationException
{
startCassandra();
executeCQLStatements(statements);
@@ -79,76 +77,74 @@ public class ThriftColumnFamilyDataTypeTest extends PigTestBase
@Test
public void testCassandraStorageDataType() throws IOException
{
- pig.registerQuery("rows = LOAD 'cassandra://thrift_ks/some_app?" + defaultParameters + "' USING CassandraStorage();");
+ pig.registerQuery("rows = LOAD 'cql://thrift_ks/some_app?" + defaultParameters + "' USING CqlNativeStorage();");
Tuple t = pig.openIterator("rows").next();
// key
assertEquals("foo", t.get(0));
// col_ascii
- Tuple column = (Tuple) t.get(1);
- assertEquals("ascii", column.get(1));
+ Object column = t.get(1);
+ assertEquals("ascii", column);
// col_bigint
- column = (Tuple) t.get(2);
- assertEquals(12345678L, column.get(1));
+ column = t.get(2);
+ assertEquals(12345678L, column);
// col_blob
- column = (Tuple) t.get(3);
- assertEquals(new DataByteArray(Hex.hexToBytes("DEADBEEF")), column.get(1));
+ column = t.get(3);
+ assertEquals(new DataByteArray(Hex.hexToBytes("DEADBEEF")), column);
// col_boolean
- column = (Tuple) t.get(4);
- assertEquals(false, column.get(1));
+ column = t.get(4);
+ assertEquals(false, column);
// col_decimal
- column = (Tuple) t.get(5);
- assertEquals("23.345", column.get(1));
+ column = t.get(5);
+ assertEquals("23.345", column);
// col_double
- column = (Tuple) t.get(6);
- assertEquals(2.7182818284590451d, column.get(1));
+ column = t.get(6);
+ assertEquals(2.7182818284590451d, column);
// col_float
- column = (Tuple) t.get(7);
- assertEquals(23.45f, column.get(1));
+ column = t.get(7);
+ assertEquals(23.45f, column);
// col_inet
- column = (Tuple) t.get(8);
- assertEquals("127.0.0.1", column.get(1));
+ column = t.get(8);
+ assertEquals("127.0.0.1", column);
// col_int
- column = (Tuple) t.get(9);
- assertEquals(23, column.get(1));
+ column = t.get(9);
+ assertEquals(23, column);
// col_text
- column = (Tuple) t.get(10);
- assertEquals("hello", column.get(1));
+ column = t.get(10);
+ assertEquals("hello", column);
// col_timestamp
- column = (Tuple) t.get(11);
- assertEquals(1296705900000L, column.get(1));
+ column = t.get(11);
+ assertEquals(1296705900000L, column);
// col_timeuuid
- column = (Tuple) t.get(12);
- assertEquals(new DataByteArray((TimeUUIDType.instance.fromString("e23f450f-53a6-11e2-7f7f-7f7f7f7f7f7f").array())), column.get(1));
+ column = t.get(12);
+ assertEquals(new DataByteArray((TimeUUIDType.instance.fromString("e23f450f-53a6-11e2-7f7f-7f7f7f7f7f7f").array())), column);
// col_uuid
- column = (Tuple) t.get(13);
- assertEquals(new DataByteArray((UUIDType.instance.fromString("550e8400-e29b-41d4-a716-446655440000").array())), column.get(1));
+ column = t.get(13);
+ assertEquals(new DataByteArray((UUIDType.instance.fromString("550e8400-e29b-41d4-a716-446655440000").array())), column);
// col_varint
- column = (Tuple) t.get(14);
- assertEquals(12345, column.get(1));
+ column = t.get(14);
+ assertEquals(12345, column);
- pig.registerQuery("cc_rows = LOAD 'cassandra://thrift_ks/cc?" + defaultParameters + "' USING CassandraStorage();");
+ pig.registerQuery("cc_rows = LOAD 'cql://thrift_ks/cc?" + defaultParameters + "' USING CqlNativeStorage();");
t = pig.openIterator("cc_rows").next();
assertEquals("chuck", t.get(0));
- DataBag columns = (DataBag) t.get(1);
- column = columns.iterator().next();
- assertEquals("kick", column.get(0));
- assertEquals(3L, column.get(1));
+ assertEquals("kick", t.get(1));
+ assertEquals(3L, t.get(2));
}
}