You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by br...@apache.org on 2012/01/24 14:45:40 UTC
git commit: Move pig to src and examples. Patch by brandonwilliams,
reviewed by Rick Branson for CASSANDRA-3713.
Updated Branches:
refs/heads/trunk eed747a6a -> dc7db0451
Move pig to src and examples.
Patch by brandonwilliams, reviewed by Rick Branson for CASSANDRA-3713.
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/dc7db045
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/dc7db045
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/dc7db045
Branch: refs/heads/trunk
Commit: dc7db045175b6bcc57dfa6ebe9eb2281e434d492
Parents: eed747a
Author: Brandon Williams <br...@apache.org>
Authored: Tue Jan 24 07:37:00 2012 -0600
Committer: Brandon Williams <br...@apache.org>
Committed: Tue Jan 24 07:37:00 2012 -0600
----------------------------------------------------------------------
build.xml | 3 +
contrib/pig/README.txt | 76 --
contrib/pig/bin/pig_cassandra | 54 --
contrib/pig/build.xml | 82 --
contrib/pig/example-script.pig | 8 -
.../cassandra/hadoop/pig/CassandraStorage.java | 668 ---------------
examples/pig/README.txt | 70 ++
examples/pig/bin/pig_cassandra | 46 +
examples/pig/example-script.pig | 8 +
.../cassandra/hadoop/pig/CassandraStorage.java | 668 +++++++++++++++
10 files changed, 795 insertions(+), 888 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/dc7db045/build.xml
----------------------------------------------------------------------
diff --git a/build.xml b/build.xml
index d9648f8..d382e34 100644
--- a/build.xml
+++ b/build.xml
@@ -378,6 +378,7 @@ url=${svn.entry.url}?pathrev=${svn.entry.commit.revision}
<exclusion groupId="commons-lang" artifactId="commons-lang"/>
</dependency>
<dependency groupId="org.apache.hadoop" artifactId="hadoop-core" version="0.20.203.0"/>
+ <dependency groupId="org.apache.pig" artifactId="pig" version="0.8.3"/>
<dependency groupId="net.sf.jopt-simple" artifactId="jopt-simple" version="3.2"/>
<dependency groupId="net.java.dev.jna" artifactId="jna" version="3.2.7"/>
@@ -420,6 +421,7 @@ url=${svn.entry.url}?pathrev=${svn.entry.commit.revision}
<dependency groupId="commons-logging" artifactId="commons-logging"/>
<dependency groupId="org.apache.rat" artifactId="apache-rat"/>
<dependency groupId="org.apache.hadoop" artifactId="hadoop-core"/>
+ <dependency groupId="org.apache.pig" artifactId="pig"/>
<dependency groupId="net.sf.jopt-simple" artifactId="jopt-simple"/>
<dependency groupId="net.java.dev.jna" artifactId="jna"/>
@@ -479,6 +481,7 @@ url=${svn.entry.url}?pathrev=${svn.entry.commit.revision}
<!-- don't need hadoop classes to run, but if you use the hadoop stuff -->
<dependency groupId="org.apache.hadoop" artifactId="hadoop-core" optional="true"/>
+ <dependency groupId="org.apache.pig" artifactId="pig" optional="true"/>
<!-- don't need jna to run, but nice to have -->
<dependency groupId="net.java.dev.jna" artifactId="jna" optional="true"/>
http://git-wip-us.apache.org/repos/asf/cassandra/blob/dc7db045/contrib/pig/README.txt
----------------------------------------------------------------------
diff --git a/contrib/pig/README.txt b/contrib/pig/README.txt
deleted file mode 100644
index 604030e..0000000
--- a/contrib/pig/README.txt
+++ /dev/null
@@ -1,76 +0,0 @@
-A Pig storage class that reads all columns from a given ColumnFamily, or writes
-properly formatted results into a ColumnFamily.
-
-Setup:
-
-First build and start a Cassandra server with the default
-configuration and set the PIG_HOME and JAVA_HOME environment
-variables to the location of a Pig >= 0.7.0 install and your Java
-install.
-
-NOTE: if you intend to _output_ to Cassandra, until there is a Pig release that
-uses jackson > 1.0.1 (see https://issues.apache.org/jira/browse/PIG-1863) you
-will need to build Pig yourself with jackson 1.4. To do this, edit Pig's
-ivy/libraries.properties, and run ant.
-
-If you would like to run using the Hadoop backend, you should
-also set PIG_CONF_DIR to the location of your Hadoop config.
-
-Finally, set the following as environment variables (uppercase,
-underscored), or as Hadoop configuration variables (lowercase, dotted):
-* PIG_INITIAL_ADDRESS or cassandra.thrift.address : initial address to connect to
-* PIG_RPC_PORT or cassandra.thrift.port : the port thrift is listening on
-* PIG_PARTITIONER or cassandra.partitioner.class : cluster partitioner
-
-For example, against a local node with the default settings, you'd use:
-export PIG_INITIAL_ADDRESS=localhost
-export PIG_RPC_PORT=9160
-export PIG_PARTITIONER=org.apache.cassandra.dht.RandomPartitioner
-
-These properties can be overridden with the following if you use different clusters
-for input and output:
-* PIG_INPUT_INITIAL_ADDRESS : initial address to connect to for reading
-* PIG_INPUT_RPC_PORT : the port thrift is listening on for reading
-* PIG_INPUT_PARTITIONER : cluster partitioner for reading
-* PIG_OUTPUT_INITIAL_ADDRESS : initial address to connect to for writing
-* PIG_OUTPUT_RPC_PORT : the port thrift is listening on for writing
-* PIG_OUTPUT_PARTITIONER : cluster partitioner for writing
-
-
-Then you can build and run it like this:
-
-contrib/pig$ ant
-contrib/pig$ bin/pig_cassandra -x local example-script.pig
-
-This will run the test script against your Cassandra instance
-and will assume that there is a MyKeyspace/MyColumnFamily with some
-data in it. It will run in local mode (see pig docs for more info).
-
-If you'd like to get to a 'grunt>' shell prompt, run:
-
-contrib/pig$ bin/pig_cassandra -x local
-
-Once the 'grunt>' shell has loaded, try a simple program like the
-following, which will determine the top 50 column names:
-
-grunt> rows = LOAD 'cassandra://MyKeyspace/MyColumnFamily' USING CassandraStorage() AS (key, columns: bag {T: tuple(name, value)});
-grunt> cols = FOREACH rows GENERATE flatten(columns);
-grunt> colnames = FOREACH cols GENERATE $0;
-grunt> namegroups = GROUP colnames BY (chararray) $0;
-grunt> namecounts = FOREACH namegroups GENERATE COUNT($1), group;
-grunt> orderednames = ORDER namecounts BY $0;
-grunt> topnames = LIMIT orderednames 50;
-grunt> dump topnames;
-
-Slices on columns can also be specified:
-grunt> rows = LOAD 'cassandra://MyKeyspace/MyColumnFamily&slice_start=C2&slice_end=C4&i&limit=1&reversed=true' USING CassandraStorage() AS (key, columns: bag {T: tuple(name, value)});
-
-Binary values for slice_start and slice_end can be escaped such as '\u0255'
-
-Outputting to Cassandra requires the same format from input, so the simplest example is:
-
-grunt> rows = LOAD 'cassandra://MyKeyspace/MyColumnFamily' USING CassandraStorage();
-grunt> STORE rows into 'cassandra://MyKeyspace/MyColumnFamily' USING CassandraStorage();
-
-Which will copy the ColumnFamily. Note that the destination ColumnFamily must
-already exist for this to work.
http://git-wip-us.apache.org/repos/asf/cassandra/blob/dc7db045/contrib/pig/bin/pig_cassandra
----------------------------------------------------------------------
diff --git a/contrib/pig/bin/pig_cassandra b/contrib/pig/bin/pig_cassandra
deleted file mode 100755
index 0dcc453..0000000
--- a/contrib/pig/bin/pig_cassandra
+++ /dev/null
@@ -1,54 +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_home="$cwd/../../../"
-
-# general jars.
-for jar in $cassandra_home/lib/*.jar $cassandra_home/build/lib/jars/*.jar; do
- CLASSPATH=$CLASSPATH:$jar
-done
-
-# cassandra_storage jar.
-STORAGE_JAR=`ls -1 $cwd/../build/*.jar`
-if [ ! -e $STORAGE_JAR ]; then
- echo "Unable to locate cassandra_storage jar: please run ant." >&2
- exit 1
-fi
-CLASSPATH=$CLASSPATH:$STORAGE_JAR
-
-if [ "x$PIG_HOME" = "x" ]; then
- echo "PIG_HOME not set: requires Pig >= 0.7.0" >&2
- exit 1
-fi
-
-# pig jar.
-for jar in $PIG_HOME/*.jar; do
- PIG_JAR=$jar
-done
-echo "Using $PIG_JAR."
-if [ ! -e $PIG_JAR ]; then
- echo "Unable to locate Pig jar" >&2
- exit 1
-fi
-
-CLASSPATH=$CLASSPATH:$PIG_JAR
-
-export PIG_CLASSPATH=$PIG_CLASSPATH:$CLASSPATH
-export PIG_OPTS="$PIG_OPTS -Dudf.import.list=org.apache.cassandra.hadoop.pig"
-cat "$cwd/../build/bootstrap.pig" - | $PIG_HOME/bin/pig $*
http://git-wip-us.apache.org/repos/asf/cassandra/blob/dc7db045/contrib/pig/build.xml
----------------------------------------------------------------------
diff --git a/contrib/pig/build.xml b/contrib/pig/build.xml
deleted file mode 100644
index 505aef1..0000000
--- a/contrib/pig/build.xml
+++ /dev/null
@@ -1,82 +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 basedir="." default="jar" name="cassandra_storage">
- <!-- stores the environment for locating PIG_HOME -->
- <property environment="env" />
- <property name="cassandra.dir" value="../.." />
- <property name="cassandra.lib" value="${cassandra.dir}/lib" />
- <property name="cassandra.classes"
- value="${cassandra.dir}/build/classes" />
- <property name="cassandra.classes.main"
- value="${cassandra.classes}/main" />
- <property name="cassandra.classes.thrift"
- value="${cassandra.classes}/thrift" />
- <property name="build.src" value="${basedir}/src" />
- <property name="build.lib" value="${basedir}/lib" />
- <property name="build.out" value="${basedir}/build" />
- <property name="build.classes" value="${build.out}/classes" />
- <property name="final.name" value="cassandra_storage" />
-
- <path id="pig.classpath">
- <fileset file="${env.PIG_HOME}/pig*.jar" />
- <fileset dir="${cassandra.lib}">
- <include name="libthrift*.jar" />
- <include name="avro*.jar" />
- </fileset>
- <fileset dir="${cassandra.dir}/build/lib/jars">
- <include name="google-collections*.jar" />
- </fileset>
- </path>
-
- <path id="classpath">
- <path refid="pig.classpath" />
- <pathelement location="${cassandra.classes.main}" />
- <pathelement location="${cassandra.classes.thrift}" />
- </path>
-
- <target name="init">
- <mkdir dir="${build.classes}" />
- </target>
-
- <target depends="init" name="build">
- <fail unless="env.PIG_HOME" message="Please set PIG_HOME to the location of a Pig >= 0.7.0 install." />
- <javac destdir="${build.classes}">
- <src path="${build.src}" />
- <classpath refid="classpath" />
- </javac>
- <!-- Build a line of jar registrations for use in the pig startup script -->
- <pathconvert pathsep="; register " property="register.line" refid="pig.classpath" />
- <echo message="register ${register.line};${line.separator}" file="${build.out}/bootstrap.pig" />
- </target>
-
- <target name="jar" depends="build">
- <mkdir dir="${build.classes}/META-INF" />
- <jar jarfile="${build.out}/${final.name}.jar">
- <fileset dir="${build.classes}" />
- <fileset dir="${cassandra.classes.main}" />
- <fileset dir="${cassandra.classes.thrift}" />
- <fileset file="${basedir}/storage-conf.xml" />
- </jar>
- </target>
-
- <target name="clean">
- <delete dir="${build.out}" />
- </target>
-</project>
http://git-wip-us.apache.org/repos/asf/cassandra/blob/dc7db045/contrib/pig/example-script.pig
----------------------------------------------------------------------
diff --git a/contrib/pig/example-script.pig b/contrib/pig/example-script.pig
deleted file mode 100644
index f88d773..0000000
--- a/contrib/pig/example-script.pig
+++ /dev/null
@@ -1,8 +0,0 @@
-rows = LOAD 'cassandra://MyKeyspace/MyColumnFamily' USING CassandraStorage() AS (key, columns: bag {T: tuple(name, value)});
-cols = FOREACH rows GENERATE flatten(columns);
-colnames = FOREACH cols GENERATE $0;
-namegroups = GROUP colnames BY (chararray) $0;
-namecounts = FOREACH namegroups GENERATE COUNT($1), group;
-orderednames = ORDER namecounts BY $0;
-topnames = LIMIT orderednames 50;
-dump topnames;
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/cassandra/blob/dc7db045/contrib/pig/src/java/org/apache/cassandra/hadoop/pig/CassandraStorage.java
----------------------------------------------------------------------
diff --git a/contrib/pig/src/java/org/apache/cassandra/hadoop/pig/CassandraStorage.java b/contrib/pig/src/java/org/apache/cassandra/hadoop/pig/CassandraStorage.java
deleted file mode 100644
index 9a84646..0000000
--- a/contrib/pig/src/java/org/apache/cassandra/hadoop/pig/CassandraStorage.java
+++ /dev/null
@@ -1,668 +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.math.BigInteger;
-import java.nio.ByteBuffer;
-import java.util.*;
-
-import org.apache.cassandra.config.ConfigurationException;
-import org.apache.cassandra.db.marshal.BytesType;
-import org.apache.cassandra.db.marshal.IntegerType;
-import org.apache.cassandra.db.marshal.TypeParser;
-import org.apache.cassandra.thrift.*;
-import org.apache.cassandra.utils.Hex;
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-
-import org.apache.cassandra.db.Column;
-import org.apache.cassandra.db.IColumn;
-import org.apache.cassandra.db.marshal.*;
-import org.apache.cassandra.hadoop.*;
-import org.apache.cassandra.thrift.Mutation;
-import org.apache.cassandra.thrift.Deletion;
-import org.apache.cassandra.thrift.ColumnOrSuperColumn;
-import org.apache.cassandra.utils.ByteBufferUtil;
-
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.mapreduce.*;
-
-import org.apache.pig.*;
-import org.apache.pig.backend.executionengine.ExecException;
-import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigSplit;
-import org.apache.pig.data.*;
-import org.apache.pig.ResourceSchema.ResourceFieldSchema;
-import org.apache.pig.impl.util.UDFContext;
-import org.apache.thrift.TDeserializer;
-import org.apache.thrift.TException;
-import org.apache.thrift.TSerializer;
-
-/**
- * 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))).
- */
-public class CassandraStorage extends LoadFunc implements StoreFuncInterface, LoadMetadata
-{
- // system environment variables that can be set to configure connection info:
- // alternatively, Hadoop JobConf variables can be set using keys from ConfigHelper
- public final static String PIG_INPUT_RPC_PORT = "PIG_INPUT_RPC_PORT";
- public final static String PIG_INPUT_INITIAL_ADDRESS = "PIG_INPUT_INITIAL_ADDRESS";
- public final static String PIG_INPUT_PARTITIONER = "PIG_INPUT_PARTITIONER";
- public final static String PIG_OUTPUT_RPC_PORT = "PIG_OUTPUT_RPC_PORT";
- public final static String PIG_OUTPUT_INITIAL_ADDRESS = "PIG_OUTPUT_INITIAL_ADDRESS";
- public final static String PIG_OUTPUT_PARTITIONER = "PIG_OUTPUT_PARTITIONER";
- public final static String PIG_RPC_PORT = "PIG_RPC_PORT";
- public final static String PIG_INITIAL_ADDRESS = "PIG_INITIAL_ADDRESS";
- public final static String PIG_PARTITIONER = "PIG_PARTITIONER";
-
- private final static ByteBuffer BOUND = ByteBufferUtil.EMPTY_BYTE_BUFFER;
- private static final Log logger = LogFactory.getLog(CassandraStorage.class);
-
- private ByteBuffer slice_start = BOUND;
- private ByteBuffer slice_end = BOUND;
- private boolean slice_reverse = false;
- private String keyspace;
- private String column_family;
- private String loadSignature;
- private String storeSignature;
-
- private Configuration conf;
- private RecordReader reader;
- private RecordWriter writer;
- private int limit;
-
- public CassandraStorage()
- {
- this(1024);
- }
-
- /**
- * @param limit: number of columns to fetch in a slice
- */
- public CassandraStorage(int limit)
- {
- super();
- this.limit = limit;
- }
-
- public int getLimit()
- {
- return limit;
- }
-
- @Override
- public Tuple getNext() throws IOException
- {
- try
- {
- // load the next pair
- if (!reader.nextKeyValue())
- return null;
-
- CfDef cfDef = getCfDef(loadSignature);
- ByteBuffer key = (ByteBuffer)reader.getCurrentKey();
- SortedMap<ByteBuffer,IColumn> cf = (SortedMap<ByteBuffer,IColumn>)reader.getCurrentValue();
- assert key != null && cf != null;
-
- // and wrap it in a tuple
- Tuple tuple = TupleFactory.getInstance().newTuple(2);
- ArrayList<Tuple> columns = new ArrayList<Tuple>();
- tuple.set(0, new DataByteArray(key.array(), key.position()+key.arrayOffset(), key.limit()+key.arrayOffset()));
- for (Map.Entry<ByteBuffer, IColumn> entry : cf.entrySet())
- {
- columns.add(columnToTuple(entry.getKey(), entry.getValue(), cfDef));
- }
-
- tuple.set(1, new DefaultDataBag(columns));
- return tuple;
- }
- catch (InterruptedException e)
- {
- throw new IOException(e.getMessage());
- }
- }
-
- private Tuple columnToTuple(ByteBuffer name, IColumn col, CfDef cfDef) throws IOException
- {
- Tuple pair = TupleFactory.getInstance().newTuple(2);
- List<AbstractType> marshallers = getDefaultMarshallers(cfDef);
- Map<ByteBuffer,AbstractType> validators = getValidatorMap(cfDef);
-
- setTupleValue(pair, 0, marshallers.get(0).compose(name));
- if (col instanceof Column)
- {
- // standard
- if (validators.get(name) == null)
- setTupleValue(pair, 1, marshallers.get(1).compose(col.value()));
- else
- setTupleValue(pair, 1, validators.get(name).compose(col.value()));
- return pair;
- }
-
- // super
- ArrayList<Tuple> subcols = new ArrayList<Tuple>();
- for (IColumn subcol : col.getSubColumns())
- subcols.add(columnToTuple(subcol.name(), subcol, cfDef));
-
- pair.set(1, new DefaultDataBag(subcols));
- return pair;
- }
-
- private void setTupleValue(Tuple pair, int position, Object value) throws ExecException
- {
- if (value instanceof BigInteger)
- pair.set(position, ((BigInteger) value).intValue());
- else if (value instanceof ByteBuffer)
- pair.set(position, new DataByteArray(ByteBufferUtil.getArray((ByteBuffer) value)));
- else
- pair.set(position, value);
- }
-
- private CfDef getCfDef(String signature)
- {
- UDFContext context = UDFContext.getUDFContext();
- Properties property = context.getUDFProperties(CassandraStorage.class);
- return cfdefFromString(property.getProperty(signature));
- }
-
- private List<AbstractType> getDefaultMarshallers(CfDef cfDef) throws IOException
- {
- ArrayList<AbstractType> marshallers = new ArrayList<AbstractType>();
- AbstractType comparator = null;
- AbstractType default_validator = null;
- AbstractType key_validator = null;
- try
- {
- comparator = TypeParser.parse(cfDef.getComparator_type());
- default_validator = TypeParser.parse(cfDef.getDefault_validation_class());
- key_validator = TypeParser.parse(cfDef.getKey_validation_class());
- }
- catch (ConfigurationException e)
- {
- throw new IOException(e);
- }
-
- marshallers.add(comparator);
- marshallers.add(default_validator);
- marshallers.add(key_validator);
- return marshallers;
- }
-
- private 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());
- validators.put(cd.name, validator);
- }
- catch (ConfigurationException e)
- {
- throw new IOException(e);
- }
- }
- }
- return validators;
- }
-
- @Override
- public InputFormat getInputFormat()
- {
- return new ColumnFamilyInputFormat();
- }
-
- @Override
- public void prepareToRead(RecordReader reader, PigSplit split)
- {
- this.reader = reader;
- }
-
- public static Map<String, String> getQueryMap(String query)
- {
- String[] params = query.split("&");
- Map<String, String> map = new HashMap<String, String>();
- for (String param : params)
- {
- String[] keyValue = param.split("=");
- map.put(keyValue[0], keyValue[1]);
- }
- return map;
- }
-
- private void setLocationFromUri(String location) throws IOException
- {
- // parse uri into keyspace and columnfamily
- String names[];
- try
- {
- if (!location.startsWith("cassandra://"))
- throw new Exception("Bad scheme.");
- 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"));
- }
- String[] parts = urlParts[0].split("/+");
- keyspace = parts[1];
- column_family = parts[2];
- }
- catch (Exception e)
- {
- throw new IOException("Expected 'cassandra://<keyspace>/<columnfamily>[?slice_start=<start>&slice_end=<end>[&reversed=true][&limit=1]]': " + e.getMessage());
- }
- }
-
- private void setConnectionInformation() throws IOException
- {
- if (System.getenv(PIG_RPC_PORT) != null)
- {
- ConfigHelper.setInputRpcPort(conf, System.getenv(PIG_RPC_PORT));
- ConfigHelper.setOutputRpcPort(conf, System.getenv(PIG_RPC_PORT));
- }
-
- if (System.getenv(PIG_INPUT_RPC_PORT) != null)
- ConfigHelper.setInputRpcPort(conf, System.getenv(PIG_INPUT_RPC_PORT));
- if (System.getenv(PIG_OUTPUT_RPC_PORT) != null)
- ConfigHelper.setOutputRpcPort(conf, System.getenv(PIG_OUTPUT_RPC_PORT));
-
- if (System.getenv(PIG_INITIAL_ADDRESS) != null)
- {
- ConfigHelper.setInputInitialAddress(conf, System.getenv(PIG_INITIAL_ADDRESS));
- ConfigHelper.setOutputInitialAddress(conf, System.getenv(PIG_INITIAL_ADDRESS));
- }
- if (System.getenv(PIG_INPUT_INITIAL_ADDRESS) != null)
- ConfigHelper.setInputInitialAddress(conf, System.getenv(PIG_INPUT_INITIAL_ADDRESS));
- if (System.getenv(PIG_OUTPUT_INITIAL_ADDRESS) != null)
- ConfigHelper.setOutputInitialAddress(conf, System.getenv(PIG_OUTPUT_INITIAL_ADDRESS));
-
- if (System.getenv(PIG_PARTITIONER) != null)
- {
- ConfigHelper.setInputPartitioner(conf, System.getenv(PIG_PARTITIONER));
- ConfigHelper.setOutputPartitioner(conf, System.getenv(PIG_PARTITIONER));
- }
- if(System.getenv(PIG_INPUT_PARTITIONER) != null)
- ConfigHelper.setInputPartitioner(conf, System.getenv(PIG_INPUT_PARTITIONER));
- if(System.getenv(PIG_OUTPUT_PARTITIONER) != null)
- ConfigHelper.setOutputPartitioner(conf, System.getenv(PIG_OUTPUT_PARTITIONER));
-
- }
-
- @Override
- public void setLocation(String location, Job job) throws IOException
- {
- conf = job.getConfiguration();
- setLocationFromUri(location);
- if (ConfigHelper.getRawInputSlicePredicate(conf) == null)
- {
- SliceRange range = new SliceRange(slice_start, slice_end, slice_reverse, limit);
- SlicePredicate predicate = new SlicePredicate().setSlice_range(range);
- ConfigHelper.setInputSlicePredicate(conf, predicate);
- }
- ConfigHelper.setInputColumnFamily(conf, keyspace, column_family);
- 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");
-
- initSchema(loadSignature);
- }
-
- public ResourceSchema getSchema(String location, Job job) throws IOException
- {
- setLocation(location, job);
- CfDef cfDef = getCfDef(loadSignature);
-
- if (cfDef.column_type.equals("Super"))
- return null;
- // top-level schema, no type
- ResourceSchema schema = new ResourceSchema();
-
- // get default marshallers and validators
- List<AbstractType> marshallers = getDefaultMarshallers(cfDef);
- Map<ByteBuffer,AbstractType> validators = getValidatorMap(cfDef);
-
- // add key
- ResourceFieldSchema keyFieldSchema = new ResourceFieldSchema();
- keyFieldSchema.setName("key");
- keyFieldSchema.setType(getPigType(marshallers.get(2)));
-
- // will become the bag of tuples
- ResourceFieldSchema bagFieldSchema = new ResourceFieldSchema();
- bagFieldSchema.setName("columns");
- bagFieldSchema.setType(DataType.BAG);
- ResourceSchema bagSchema = new ResourceSchema();
-
- List<ResourceFieldSchema> tupleFields = new ArrayList<ResourceFieldSchema>();
-
- // default comparator/validator
- ResourceSchema innerTupleSchema = new ResourceSchema();
- ResourceFieldSchema tupleField = new ResourceFieldSchema();
- tupleField.setType(DataType.TUPLE);
- tupleField.setSchema(innerTupleSchema);
-
- ResourceFieldSchema colSchema = new ResourceFieldSchema();
- colSchema.setName("name");
- colSchema.setType(getPigType(marshallers.get(0)));
- tupleFields.add(colSchema);
-
- ResourceFieldSchema valSchema = new ResourceFieldSchema();
- AbstractType validator = marshallers.get(1);
- valSchema.setName("value");
- valSchema.setType(getPigType(validator));
- tupleFields.add(valSchema);
-
- // defined validators/indexes
- for (ColumnDef cdef : cfDef.column_metadata)
- {
- colSchema = new ResourceFieldSchema();
- colSchema.setName(new String(cdef.getName()));
- colSchema.setType(getPigType(marshallers.get(0)));
- tupleFields.add(colSchema);
-
- valSchema = new ResourceFieldSchema();
- validator = validators.get(cdef.getName());
- if (validator == null)
- validator = marshallers.get(1);
- valSchema.setName("value");
- valSchema.setType(getPigType(validator));
- tupleFields.add(valSchema);
- }
- innerTupleSchema.setFields(tupleFields.toArray(new ResourceFieldSchema[tupleFields.size()]));
-
- // a bag can contain only one tuple, but that tuple can contain anything
- bagSchema.setFields(new ResourceFieldSchema[] { tupleField });
- bagFieldSchema.setSchema(bagSchema);
- // top level schema contains everything
- schema.setFields(new ResourceFieldSchema[] { keyFieldSchema, bagFieldSchema });
- return schema;
- }
-
- private byte getPigType(AbstractType type)
- {
- if (type instanceof LongType)
- return DataType.LONG;
- else if (type instanceof IntegerType)
- return DataType.INTEGER;
- else if (type instanceof AsciiType)
- return DataType.CHARARRAY;
- else if (type instanceof UTF8Type)
- return DataType.CHARARRAY;
- else if (type instanceof FloatType)
- return DataType.FLOAT;
- else if (type instanceof DoubleType)
- return DataType.DOUBLE;
- return DataType.BYTEARRAY;
- }
-
- public ResourceStatistics getStatistics(String location, Job job)
- {
- return null;
- }
-
- public String[] getPartitionKeys(String location, Job job)
- {
- return null;
- }
-
- public void setPartitionFilter(Expression partitionFilter)
- {
- // no-op
- }
-
- @Override
- public String relativeToAbsolutePath(String location, Path curDir) throws IOException
- {
- return location;
- }
-
- @Override
- public void setUDFContextSignature(String signature)
- {
- this.loadSignature = signature;
- }
-
- /* StoreFunc methods */
- public void setStoreFuncUDFContextSignature(String signature)
- {
- this.storeSignature = signature;
- }
-
- public String relToAbsPathForStoreLocation(String location, Path curDir) throws IOException
- {
- return relativeToAbsolutePath(location, curDir);
- }
-
- public void setStoreLocation(String location, Job job) throws IOException
- {
- conf = job.getConfiguration();
- setLocationFromUri(location);
- 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");
-
- initSchema(storeSignature);
- }
-
- public OutputFormat getOutputFormat()
- {
- return new ColumnFamilyOutputFormat();
- }
-
- public void checkSchema(ResourceSchema schema) throws IOException
- {
- // we don't care about types, they all get casted to ByteBuffers
- }
-
- public void prepareToWrite(RecordWriter writer)
- {
- this.writer = writer;
- }
-
- private ByteBuffer objToBB(Object o)
- {
- if (o == null)
- return (ByteBuffer)o;
- if (o instanceof java.lang.String)
- o = new DataByteArray((String)o);
- return ByteBuffer.wrap(((DataByteArray) o).get());
- }
-
- public void putNext(Tuple t) throws ExecException, IOException
- {
- ByteBuffer key = objToBB(t.get(0));
- DefaultDataBag pairs = (DefaultDataBag) t.get(1);
- ArrayList<Mutation> mutationList = new ArrayList<Mutation>();
- CfDef cfDef = getCfDef(storeSignature);
- try
- {
- for (Tuple pair : pairs)
- {
- Mutation mutation = new Mutation();
- if (DataType.findType(pair.get(1)) == DataType.BAG) // supercolumn
- {
- org.apache.cassandra.thrift.SuperColumn sc = new org.apache.cassandra.thrift.SuperColumn();
- sc.name = objToBB(pair.get(0));
- ArrayList<org.apache.cassandra.thrift.Column> columns = new ArrayList<org.apache.cassandra.thrift.Column>();
- for (Tuple subcol : (DefaultDataBag) pair.get(1))
- {
- org.apache.cassandra.thrift.Column column = new org.apache.cassandra.thrift.Column();
- column.name = objToBB(subcol.get(0));
- column.value = objToBB(subcol.get(1));
- column.setTimestamp(System.currentTimeMillis() * 1000);
- columns.add(column);
- }
- if (columns.isEmpty()) // a deletion
- {
- mutation.deletion = new Deletion();
- mutation.deletion.super_column = objToBB(pair.get(0));
- mutation.deletion.setTimestamp(System.currentTimeMillis() * 1000);
- }
- else
- {
- sc.columns = columns;
- mutation.column_or_supercolumn = new ColumnOrSuperColumn();
- mutation.column_or_supercolumn.super_column = sc;
- }
- }
- else // assume column since it couldn't be anything else
- {
- if (pair.get(1) == null)
- {
- mutation.deletion = new Deletion();
- mutation.deletion.predicate = new org.apache.cassandra.thrift.SlicePredicate();
- mutation.deletion.predicate.column_names = Arrays.asList(objToBB(pair.get(0)));
- mutation.deletion.setTimestamp(System.currentTimeMillis() * 1000);
- }
- else
- {
- org.apache.cassandra.thrift.Column column = new org.apache.cassandra.thrift.Column();
- column.name = objToBB(pair.get(0));
- column.value = objToBB(pair.get(1));
- column.setTimestamp(System.currentTimeMillis() * 1000);
- mutation.column_or_supercolumn = new ColumnOrSuperColumn();
- mutation.column_or_supercolumn.column = column;
- }
- }
- mutationList.add(mutation);
- }
- }
- catch (ClassCastException e)
- {
- throw new IOException(e + " Output must be (key, {(column,value)...}) for ColumnFamily or (key, {supercolumn:{(column,value)...}...}) for SuperColumnFamily", e);
- }
- try
- {
- writer.write(key, mutationList);
- }
- catch (InterruptedException e)
- {
- throw new IOException(e);
- }
- }
-
- public void cleanupOnFailure(String failure, Job job)
- {
- }
-
- /* Methods to get the column family schema from Cassandra */
-
- private void initSchema(String signature)
- {
- UDFContext context = UDFContext.getUDFContext();
- Properties property = context.getUDFProperties(CassandraStorage.class);
-
- // Only get the schema if we haven't already gotten it
- if (!property.containsKey(signature))
- {
- Cassandra.Client client = null;
- try
- {
- client = ConfigHelper.getClientFromInputAddressList(conf);
- CfDef cfDef = null;
- client.set_keyspace(keyspace);
- KsDef ksDef = client.describe_keyspace(keyspace);
- List<CfDef> defs = ksDef.getCf_defs();
- for (CfDef def : defs)
- {
- if (column_family.equalsIgnoreCase(def.getName()))
- {
- cfDef = def;
- break;
- }
- }
- property.setProperty(signature, cfdefToString(cfDef));
- }
- catch (TException e)
- {
- throw new RuntimeException(e);
- }
- catch (InvalidRequestException e)
- {
- throw new RuntimeException(e);
- }
- catch (NotFoundException e)
- {
- throw new RuntimeException(e);
- }
- catch (IOException e)
- {
- throw new RuntimeException(e);
- }
- }
- }
-
- private static String cfdefToString(CfDef cfDef)
- {
- 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 RuntimeException(e);
- }
- }
-
- private static CfDef cfdefFromString(String st)
- {
- 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 RuntimeException(e);
- }
- return cfDef;
- }
-}
-
http://git-wip-us.apache.org/repos/asf/cassandra/blob/dc7db045/examples/pig/README.txt
----------------------------------------------------------------------
diff --git a/examples/pig/README.txt b/examples/pig/README.txt
new file mode 100644
index 0000000..aea0a77
--- /dev/null
+++ b/examples/pig/README.txt
@@ -0,0 +1,70 @@
+A Pig storage class that reads all columns from a given ColumnFamily, or writes
+properly formatted results into a ColumnFamily.
+
+Setup:
+
+First build and start a Cassandra server with the default
+configuration and set the PIG_HOME and JAVA_HOME environment
+variables to the location of a Pig >= 0.7.0 install and your Java
+install.
+
+If you would like to run using the Hadoop backend, you should
+also set PIG_CONF_DIR to the location of your Hadoop config.
+
+Finally, set the following as environment variables (uppercase,
+underscored), or as Hadoop configuration variables (lowercase, dotted):
+* PIG_INITIAL_ADDRESS or cassandra.thrift.address : initial address to connect to
+* PIG_RPC_PORT or cassandra.thrift.port : the port thrift is listening on
+* PIG_PARTITIONER or cassandra.partitioner.class : cluster partitioner
+
+For example, against a local node with the default settings, you'd use:
+export PIG_INITIAL_ADDRESS=localhost
+export PIG_RPC_PORT=9160
+export PIG_PARTITIONER=org.apache.cassandra.dht.RandomPartitioner
+
+These properties can be overridden with the following if you use different clusters
+for input and output:
+* PIG_INPUT_INITIAL_ADDRESS : initial address to connect to for reading
+* PIG_INPUT_RPC_PORT : the port thrift is listening on for reading
+* PIG_INPUT_PARTITIONER : cluster partitioner for reading
+* PIG_OUTPUT_INITIAL_ADDRESS : initial address to connect to for writing
+* PIG_OUTPUT_RPC_PORT : the port thrift is listening on for writing
+* PIG_OUTPUT_PARTITIONER : cluster partitioner for writing
+
+
+Then you can run it like this:
+
+examples/pig$ bin/pig_cassandra -x local example-script.pig
+
+This will run the test script against your Cassandra instance
+and will assume that there is a MyKeyspace/MyColumnFamily with some
+data in it. It will run in local mode (see pig docs for more info).
+
+If you'd like to get to a 'grunt>' shell prompt, run:
+
+examples/pig$ bin/pig_cassandra -x local
+
+Once the 'grunt>' shell has loaded, try a simple program like the
+following, which will determine the top 50 column names:
+
+grunt> rows = LOAD 'cassandra://MyKeyspace/MyColumnFamily' USING CassandraStorage() AS (key, columns: bag {T: tuple(name, value)});
+grunt> cols = FOREACH rows GENERATE flatten(columns);
+grunt> colnames = FOREACH cols GENERATE $0;
+grunt> namegroups = GROUP colnames BY (chararray) $0;
+grunt> namecounts = FOREACH namegroups GENERATE COUNT($1), group;
+grunt> orderednames = ORDER namecounts BY $0;
+grunt> topnames = LIMIT orderednames 50;
+grunt> dump topnames;
+
+Slices on columns can also be specified:
+grunt> rows = LOAD 'cassandra://MyKeyspace/MyColumnFamily&slice_start=C2&slice_end=C4&limit=1&reversed=true' USING CassandraStorage() AS (key, columns: bag {T: tuple(name, value)});
+
+Binary values for slice_start and slice_end can be escaped such as '\u0255'
+
+Outputting to Cassandra requires the same format from input, so the simplest example is:
+
+grunt> rows = LOAD 'cassandra://MyKeyspace/MyColumnFamily' USING CassandraStorage();
+grunt> STORE rows into 'cassandra://MyKeyspace/MyColumnFamily' USING CassandraStorage();
+
+Which will copy the ColumnFamily. Note that the destination ColumnFamily must
+already exist for this to work.
http://git-wip-us.apache.org/repos/asf/cassandra/blob/dc7db045/examples/pig/bin/pig_cassandra
----------------------------------------------------------------------
diff --git a/examples/pig/bin/pig_cassandra b/examples/pig/bin/pig_cassandra
new file mode 100644
index 0000000..5d98888
--- /dev/null
+++ b/examples/pig/bin/pig_cassandra
@@ -0,0 +1,46 @@
+#!/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_home="$cwd/../../../"
+
+# general jars.
+for jar in $cassandra_home/lib/*.jar $cassandra_home/build/lib/jars/*.jar $cassandra_home/build/apache-cassandra*.jar; do
+ CLASSPATH=$CLASSPATH:$jar
+done
+
+if [ "x$PIG_HOME" = "x" ]; then
+ echo "PIG_HOME not set: requires Pig >= 0.7.0" >&2
+ exit 1
+fi
+
+# pig jar.
+for jar in $PIG_HOME/*.jar; do
+ PIG_JAR=$jar
+done
+echo "Using $PIG_JAR."
+if [ ! -e $PIG_JAR ]; then
+ echo "Unable to locate Pig jar" >&2
+ exit 1
+fi
+
+CLASSPATH=$CLASSPATH:$PIG_JAR
+
+export PIG_CLASSPATH=$PIG_CLASSPATH:$CLASSPATH
+export PIG_OPTS="$PIG_OPTS -Dudf.import.list=org.apache.cassandra.hadoop.pig"
+$PIG_HOME/bin/pig $*
http://git-wip-us.apache.org/repos/asf/cassandra/blob/dc7db045/examples/pig/example-script.pig
----------------------------------------------------------------------
diff --git a/examples/pig/example-script.pig b/examples/pig/example-script.pig
new file mode 100644
index 0000000..f88d773
--- /dev/null
+++ b/examples/pig/example-script.pig
@@ -0,0 +1,8 @@
+rows = LOAD 'cassandra://MyKeyspace/MyColumnFamily' USING CassandraStorage() AS (key, columns: bag {T: tuple(name, value)});
+cols = FOREACH rows GENERATE flatten(columns);
+colnames = FOREACH cols GENERATE $0;
+namegroups = GROUP colnames BY (chararray) $0;
+namecounts = FOREACH namegroups GENERATE COUNT($1), group;
+orderednames = ORDER namecounts BY $0;
+topnames = LIMIT orderednames 50;
+dump topnames;
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/cassandra/blob/dc7db045/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
new file mode 100644
index 0000000..9a84646
--- /dev/null
+++ b/src/java/org/apache/cassandra/hadoop/pig/CassandraStorage.java
@@ -0,0 +1,668 @@
+/**
+ * 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.math.BigInteger;
+import java.nio.ByteBuffer;
+import java.util.*;
+
+import org.apache.cassandra.config.ConfigurationException;
+import org.apache.cassandra.db.marshal.BytesType;
+import org.apache.cassandra.db.marshal.IntegerType;
+import org.apache.cassandra.db.marshal.TypeParser;
+import org.apache.cassandra.thrift.*;
+import org.apache.cassandra.utils.Hex;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import org.apache.cassandra.db.Column;
+import org.apache.cassandra.db.IColumn;
+import org.apache.cassandra.db.marshal.*;
+import org.apache.cassandra.hadoop.*;
+import org.apache.cassandra.thrift.Mutation;
+import org.apache.cassandra.thrift.Deletion;
+import org.apache.cassandra.thrift.ColumnOrSuperColumn;
+import org.apache.cassandra.utils.ByteBufferUtil;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.mapreduce.*;
+
+import org.apache.pig.*;
+import org.apache.pig.backend.executionengine.ExecException;
+import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigSplit;
+import org.apache.pig.data.*;
+import org.apache.pig.ResourceSchema.ResourceFieldSchema;
+import org.apache.pig.impl.util.UDFContext;
+import org.apache.thrift.TDeserializer;
+import org.apache.thrift.TException;
+import org.apache.thrift.TSerializer;
+
+/**
+ * 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))).
+ */
+public class CassandraStorage extends LoadFunc implements StoreFuncInterface, LoadMetadata
+{
+ // system environment variables that can be set to configure connection info:
+ // alternatively, Hadoop JobConf variables can be set using keys from ConfigHelper
+ public final static String PIG_INPUT_RPC_PORT = "PIG_INPUT_RPC_PORT";
+ public final static String PIG_INPUT_INITIAL_ADDRESS = "PIG_INPUT_INITIAL_ADDRESS";
+ public final static String PIG_INPUT_PARTITIONER = "PIG_INPUT_PARTITIONER";
+ public final static String PIG_OUTPUT_RPC_PORT = "PIG_OUTPUT_RPC_PORT";
+ public final static String PIG_OUTPUT_INITIAL_ADDRESS = "PIG_OUTPUT_INITIAL_ADDRESS";
+ public final static String PIG_OUTPUT_PARTITIONER = "PIG_OUTPUT_PARTITIONER";
+ public final static String PIG_RPC_PORT = "PIG_RPC_PORT";
+ public final static String PIG_INITIAL_ADDRESS = "PIG_INITIAL_ADDRESS";
+ public final static String PIG_PARTITIONER = "PIG_PARTITIONER";
+
+ private final static ByteBuffer BOUND = ByteBufferUtil.EMPTY_BYTE_BUFFER;
+ private static final Log logger = LogFactory.getLog(CassandraStorage.class);
+
+ private ByteBuffer slice_start = BOUND;
+ private ByteBuffer slice_end = BOUND;
+ private boolean slice_reverse = false;
+ private String keyspace;
+ private String column_family;
+ private String loadSignature;
+ private String storeSignature;
+
+ private Configuration conf;
+ private RecordReader reader;
+ private RecordWriter writer;
+ private int limit;
+
+ public CassandraStorage()
+ {
+ this(1024);
+ }
+
+ /**
+ * @param limit: number of columns to fetch in a slice
+ */
+ public CassandraStorage(int limit)
+ {
+ super();
+ this.limit = limit;
+ }
+
+ public int getLimit()
+ {
+ return limit;
+ }
+
+ @Override
+ public Tuple getNext() throws IOException
+ {
+ try
+ {
+ // load the next pair
+ if (!reader.nextKeyValue())
+ return null;
+
+ CfDef cfDef = getCfDef(loadSignature);
+ ByteBuffer key = (ByteBuffer)reader.getCurrentKey();
+ SortedMap<ByteBuffer,IColumn> cf = (SortedMap<ByteBuffer,IColumn>)reader.getCurrentValue();
+ assert key != null && cf != null;
+
+ // and wrap it in a tuple
+ Tuple tuple = TupleFactory.getInstance().newTuple(2);
+ ArrayList<Tuple> columns = new ArrayList<Tuple>();
+ tuple.set(0, new DataByteArray(key.array(), key.position()+key.arrayOffset(), key.limit()+key.arrayOffset()));
+ for (Map.Entry<ByteBuffer, IColumn> entry : cf.entrySet())
+ {
+ columns.add(columnToTuple(entry.getKey(), entry.getValue(), cfDef));
+ }
+
+ tuple.set(1, new DefaultDataBag(columns));
+ return tuple;
+ }
+ catch (InterruptedException e)
+ {
+ throw new IOException(e.getMessage());
+ }
+ }
+
+ private Tuple columnToTuple(ByteBuffer name, IColumn col, CfDef cfDef) throws IOException
+ {
+ Tuple pair = TupleFactory.getInstance().newTuple(2);
+ List<AbstractType> marshallers = getDefaultMarshallers(cfDef);
+ Map<ByteBuffer,AbstractType> validators = getValidatorMap(cfDef);
+
+ setTupleValue(pair, 0, marshallers.get(0).compose(name));
+ if (col instanceof Column)
+ {
+ // standard
+ if (validators.get(name) == null)
+ setTupleValue(pair, 1, marshallers.get(1).compose(col.value()));
+ else
+ setTupleValue(pair, 1, validators.get(name).compose(col.value()));
+ return pair;
+ }
+
+ // super
+ ArrayList<Tuple> subcols = new ArrayList<Tuple>();
+ for (IColumn subcol : col.getSubColumns())
+ subcols.add(columnToTuple(subcol.name(), subcol, cfDef));
+
+ pair.set(1, new DefaultDataBag(subcols));
+ return pair;
+ }
+
+ private void setTupleValue(Tuple pair, int position, Object value) throws ExecException
+ {
+ if (value instanceof BigInteger)
+ pair.set(position, ((BigInteger) value).intValue());
+ else if (value instanceof ByteBuffer)
+ pair.set(position, new DataByteArray(ByteBufferUtil.getArray((ByteBuffer) value)));
+ else
+ pair.set(position, value);
+ }
+
+ private CfDef getCfDef(String signature)
+ {
+ UDFContext context = UDFContext.getUDFContext();
+ Properties property = context.getUDFProperties(CassandraStorage.class);
+ return cfdefFromString(property.getProperty(signature));
+ }
+
+ private List<AbstractType> getDefaultMarshallers(CfDef cfDef) throws IOException
+ {
+ ArrayList<AbstractType> marshallers = new ArrayList<AbstractType>();
+ AbstractType comparator = null;
+ AbstractType default_validator = null;
+ AbstractType key_validator = null;
+ try
+ {
+ comparator = TypeParser.parse(cfDef.getComparator_type());
+ default_validator = TypeParser.parse(cfDef.getDefault_validation_class());
+ key_validator = TypeParser.parse(cfDef.getKey_validation_class());
+ }
+ catch (ConfigurationException e)
+ {
+ throw new IOException(e);
+ }
+
+ marshallers.add(comparator);
+ marshallers.add(default_validator);
+ marshallers.add(key_validator);
+ return marshallers;
+ }
+
+ private 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());
+ validators.put(cd.name, validator);
+ }
+ catch (ConfigurationException e)
+ {
+ throw new IOException(e);
+ }
+ }
+ }
+ return validators;
+ }
+
+ @Override
+ public InputFormat getInputFormat()
+ {
+ return new ColumnFamilyInputFormat();
+ }
+
+ @Override
+ public void prepareToRead(RecordReader reader, PigSplit split)
+ {
+ this.reader = reader;
+ }
+
+ public static Map<String, String> getQueryMap(String query)
+ {
+ String[] params = query.split("&");
+ Map<String, String> map = new HashMap<String, String>();
+ for (String param : params)
+ {
+ String[] keyValue = param.split("=");
+ map.put(keyValue[0], keyValue[1]);
+ }
+ return map;
+ }
+
+ private void setLocationFromUri(String location) throws IOException
+ {
+ // parse uri into keyspace and columnfamily
+ String names[];
+ try
+ {
+ if (!location.startsWith("cassandra://"))
+ throw new Exception("Bad scheme.");
+ 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"));
+ }
+ String[] parts = urlParts[0].split("/+");
+ keyspace = parts[1];
+ column_family = parts[2];
+ }
+ catch (Exception e)
+ {
+ throw new IOException("Expected 'cassandra://<keyspace>/<columnfamily>[?slice_start=<start>&slice_end=<end>[&reversed=true][&limit=1]]': " + e.getMessage());
+ }
+ }
+
+ private void setConnectionInformation() throws IOException
+ {
+ if (System.getenv(PIG_RPC_PORT) != null)
+ {
+ ConfigHelper.setInputRpcPort(conf, System.getenv(PIG_RPC_PORT));
+ ConfigHelper.setOutputRpcPort(conf, System.getenv(PIG_RPC_PORT));
+ }
+
+ if (System.getenv(PIG_INPUT_RPC_PORT) != null)
+ ConfigHelper.setInputRpcPort(conf, System.getenv(PIG_INPUT_RPC_PORT));
+ if (System.getenv(PIG_OUTPUT_RPC_PORT) != null)
+ ConfigHelper.setOutputRpcPort(conf, System.getenv(PIG_OUTPUT_RPC_PORT));
+
+ if (System.getenv(PIG_INITIAL_ADDRESS) != null)
+ {
+ ConfigHelper.setInputInitialAddress(conf, System.getenv(PIG_INITIAL_ADDRESS));
+ ConfigHelper.setOutputInitialAddress(conf, System.getenv(PIG_INITIAL_ADDRESS));
+ }
+ if (System.getenv(PIG_INPUT_INITIAL_ADDRESS) != null)
+ ConfigHelper.setInputInitialAddress(conf, System.getenv(PIG_INPUT_INITIAL_ADDRESS));
+ if (System.getenv(PIG_OUTPUT_INITIAL_ADDRESS) != null)
+ ConfigHelper.setOutputInitialAddress(conf, System.getenv(PIG_OUTPUT_INITIAL_ADDRESS));
+
+ if (System.getenv(PIG_PARTITIONER) != null)
+ {
+ ConfigHelper.setInputPartitioner(conf, System.getenv(PIG_PARTITIONER));
+ ConfigHelper.setOutputPartitioner(conf, System.getenv(PIG_PARTITIONER));
+ }
+ if(System.getenv(PIG_INPUT_PARTITIONER) != null)
+ ConfigHelper.setInputPartitioner(conf, System.getenv(PIG_INPUT_PARTITIONER));
+ if(System.getenv(PIG_OUTPUT_PARTITIONER) != null)
+ ConfigHelper.setOutputPartitioner(conf, System.getenv(PIG_OUTPUT_PARTITIONER));
+
+ }
+
+ @Override
+ public void setLocation(String location, Job job) throws IOException
+ {
+ conf = job.getConfiguration();
+ setLocationFromUri(location);
+ if (ConfigHelper.getRawInputSlicePredicate(conf) == null)
+ {
+ SliceRange range = new SliceRange(slice_start, slice_end, slice_reverse, limit);
+ SlicePredicate predicate = new SlicePredicate().setSlice_range(range);
+ ConfigHelper.setInputSlicePredicate(conf, predicate);
+ }
+ ConfigHelper.setInputColumnFamily(conf, keyspace, column_family);
+ 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");
+
+ initSchema(loadSignature);
+ }
+
+ public ResourceSchema getSchema(String location, Job job) throws IOException
+ {
+ setLocation(location, job);
+ CfDef cfDef = getCfDef(loadSignature);
+
+ if (cfDef.column_type.equals("Super"))
+ return null;
+ // top-level schema, no type
+ ResourceSchema schema = new ResourceSchema();
+
+ // get default marshallers and validators
+ List<AbstractType> marshallers = getDefaultMarshallers(cfDef);
+ Map<ByteBuffer,AbstractType> validators = getValidatorMap(cfDef);
+
+ // add key
+ ResourceFieldSchema keyFieldSchema = new ResourceFieldSchema();
+ keyFieldSchema.setName("key");
+ keyFieldSchema.setType(getPigType(marshallers.get(2)));
+
+ // will become the bag of tuples
+ ResourceFieldSchema bagFieldSchema = new ResourceFieldSchema();
+ bagFieldSchema.setName("columns");
+ bagFieldSchema.setType(DataType.BAG);
+ ResourceSchema bagSchema = new ResourceSchema();
+
+ List<ResourceFieldSchema> tupleFields = new ArrayList<ResourceFieldSchema>();
+
+ // default comparator/validator
+ ResourceSchema innerTupleSchema = new ResourceSchema();
+ ResourceFieldSchema tupleField = new ResourceFieldSchema();
+ tupleField.setType(DataType.TUPLE);
+ tupleField.setSchema(innerTupleSchema);
+
+ ResourceFieldSchema colSchema = new ResourceFieldSchema();
+ colSchema.setName("name");
+ colSchema.setType(getPigType(marshallers.get(0)));
+ tupleFields.add(colSchema);
+
+ ResourceFieldSchema valSchema = new ResourceFieldSchema();
+ AbstractType validator = marshallers.get(1);
+ valSchema.setName("value");
+ valSchema.setType(getPigType(validator));
+ tupleFields.add(valSchema);
+
+ // defined validators/indexes
+ for (ColumnDef cdef : cfDef.column_metadata)
+ {
+ colSchema = new ResourceFieldSchema();
+ colSchema.setName(new String(cdef.getName()));
+ colSchema.setType(getPigType(marshallers.get(0)));
+ tupleFields.add(colSchema);
+
+ valSchema = new ResourceFieldSchema();
+ validator = validators.get(cdef.getName());
+ if (validator == null)
+ validator = marshallers.get(1);
+ valSchema.setName("value");
+ valSchema.setType(getPigType(validator));
+ tupleFields.add(valSchema);
+ }
+ innerTupleSchema.setFields(tupleFields.toArray(new ResourceFieldSchema[tupleFields.size()]));
+
+ // a bag can contain only one tuple, but that tuple can contain anything
+ bagSchema.setFields(new ResourceFieldSchema[] { tupleField });
+ bagFieldSchema.setSchema(bagSchema);
+ // top level schema contains everything
+ schema.setFields(new ResourceFieldSchema[] { keyFieldSchema, bagFieldSchema });
+ return schema;
+ }
+
+ private byte getPigType(AbstractType type)
+ {
+ if (type instanceof LongType)
+ return DataType.LONG;
+ else if (type instanceof IntegerType)
+ return DataType.INTEGER;
+ else if (type instanceof AsciiType)
+ return DataType.CHARARRAY;
+ else if (type instanceof UTF8Type)
+ return DataType.CHARARRAY;
+ else if (type instanceof FloatType)
+ return DataType.FLOAT;
+ else if (type instanceof DoubleType)
+ return DataType.DOUBLE;
+ return DataType.BYTEARRAY;
+ }
+
+ public ResourceStatistics getStatistics(String location, Job job)
+ {
+ return null;
+ }
+
+ public String[] getPartitionKeys(String location, Job job)
+ {
+ return null;
+ }
+
+ public void setPartitionFilter(Expression partitionFilter)
+ {
+ // no-op
+ }
+
+ @Override
+ public String relativeToAbsolutePath(String location, Path curDir) throws IOException
+ {
+ return location;
+ }
+
+ @Override
+ public void setUDFContextSignature(String signature)
+ {
+ this.loadSignature = signature;
+ }
+
+ /* StoreFunc methods */
+ public void setStoreFuncUDFContextSignature(String signature)
+ {
+ this.storeSignature = signature;
+ }
+
+ public String relToAbsPathForStoreLocation(String location, Path curDir) throws IOException
+ {
+ return relativeToAbsolutePath(location, curDir);
+ }
+
+ public void setStoreLocation(String location, Job job) throws IOException
+ {
+ conf = job.getConfiguration();
+ setLocationFromUri(location);
+ 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");
+
+ initSchema(storeSignature);
+ }
+
+ public OutputFormat getOutputFormat()
+ {
+ return new ColumnFamilyOutputFormat();
+ }
+
+ public void checkSchema(ResourceSchema schema) throws IOException
+ {
+ // we don't care about types, they all get casted to ByteBuffers
+ }
+
+ public void prepareToWrite(RecordWriter writer)
+ {
+ this.writer = writer;
+ }
+
+ private ByteBuffer objToBB(Object o)
+ {
+ if (o == null)
+ return (ByteBuffer)o;
+ if (o instanceof java.lang.String)
+ o = new DataByteArray((String)o);
+ return ByteBuffer.wrap(((DataByteArray) o).get());
+ }
+
+ public void putNext(Tuple t) throws ExecException, IOException
+ {
+ ByteBuffer key = objToBB(t.get(0));
+ DefaultDataBag pairs = (DefaultDataBag) t.get(1);
+ ArrayList<Mutation> mutationList = new ArrayList<Mutation>();
+ CfDef cfDef = getCfDef(storeSignature);
+ try
+ {
+ for (Tuple pair : pairs)
+ {
+ Mutation mutation = new Mutation();
+ if (DataType.findType(pair.get(1)) == DataType.BAG) // supercolumn
+ {
+ org.apache.cassandra.thrift.SuperColumn sc = new org.apache.cassandra.thrift.SuperColumn();
+ sc.name = objToBB(pair.get(0));
+ ArrayList<org.apache.cassandra.thrift.Column> columns = new ArrayList<org.apache.cassandra.thrift.Column>();
+ for (Tuple subcol : (DefaultDataBag) pair.get(1))
+ {
+ org.apache.cassandra.thrift.Column column = new org.apache.cassandra.thrift.Column();
+ column.name = objToBB(subcol.get(0));
+ column.value = objToBB(subcol.get(1));
+ column.setTimestamp(System.currentTimeMillis() * 1000);
+ columns.add(column);
+ }
+ if (columns.isEmpty()) // a deletion
+ {
+ mutation.deletion = new Deletion();
+ mutation.deletion.super_column = objToBB(pair.get(0));
+ mutation.deletion.setTimestamp(System.currentTimeMillis() * 1000);
+ }
+ else
+ {
+ sc.columns = columns;
+ mutation.column_or_supercolumn = new ColumnOrSuperColumn();
+ mutation.column_or_supercolumn.super_column = sc;
+ }
+ }
+ else // assume column since it couldn't be anything else
+ {
+ if (pair.get(1) == null)
+ {
+ mutation.deletion = new Deletion();
+ mutation.deletion.predicate = new org.apache.cassandra.thrift.SlicePredicate();
+ mutation.deletion.predicate.column_names = Arrays.asList(objToBB(pair.get(0)));
+ mutation.deletion.setTimestamp(System.currentTimeMillis() * 1000);
+ }
+ else
+ {
+ org.apache.cassandra.thrift.Column column = new org.apache.cassandra.thrift.Column();
+ column.name = objToBB(pair.get(0));
+ column.value = objToBB(pair.get(1));
+ column.setTimestamp(System.currentTimeMillis() * 1000);
+ mutation.column_or_supercolumn = new ColumnOrSuperColumn();
+ mutation.column_or_supercolumn.column = column;
+ }
+ }
+ mutationList.add(mutation);
+ }
+ }
+ catch (ClassCastException e)
+ {
+ throw new IOException(e + " Output must be (key, {(column,value)...}) for ColumnFamily or (key, {supercolumn:{(column,value)...}...}) for SuperColumnFamily", e);
+ }
+ try
+ {
+ writer.write(key, mutationList);
+ }
+ catch (InterruptedException e)
+ {
+ throw new IOException(e);
+ }
+ }
+
+ public void cleanupOnFailure(String failure, Job job)
+ {
+ }
+
+ /* Methods to get the column family schema from Cassandra */
+
+ private void initSchema(String signature)
+ {
+ UDFContext context = UDFContext.getUDFContext();
+ Properties property = context.getUDFProperties(CassandraStorage.class);
+
+ // Only get the schema if we haven't already gotten it
+ if (!property.containsKey(signature))
+ {
+ Cassandra.Client client = null;
+ try
+ {
+ client = ConfigHelper.getClientFromInputAddressList(conf);
+ CfDef cfDef = null;
+ client.set_keyspace(keyspace);
+ KsDef ksDef = client.describe_keyspace(keyspace);
+ List<CfDef> defs = ksDef.getCf_defs();
+ for (CfDef def : defs)
+ {
+ if (column_family.equalsIgnoreCase(def.getName()))
+ {
+ cfDef = def;
+ break;
+ }
+ }
+ property.setProperty(signature, cfdefToString(cfDef));
+ }
+ catch (TException e)
+ {
+ throw new RuntimeException(e);
+ }
+ catch (InvalidRequestException e)
+ {
+ throw new RuntimeException(e);
+ }
+ catch (NotFoundException e)
+ {
+ throw new RuntimeException(e);
+ }
+ catch (IOException e)
+ {
+ throw new RuntimeException(e);
+ }
+ }
+ }
+
+ private static String cfdefToString(CfDef cfDef)
+ {
+ 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 RuntimeException(e);
+ }
+ }
+
+ private static CfDef cfdefFromString(String st)
+ {
+ 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 RuntimeException(e);
+ }
+ return cfDef;
+ }
+}
+