diff options
author | Marcelo Vanzin <vanzin@cloudera.com> | 2016-04-25 10:20:51 -0700 |
---|---|---|
committer | Marcelo Vanzin <vanzin@cloudera.com> | 2016-04-25 10:20:51 -0700 |
commit | a680562a6f87a03a00f71bad1c424267ae75c641 (patch) | |
tree | fe08025da6437124c283e9450dd3bbc0e99c411d /examples/src/main/scala | |
parent | bfda09991398ce44be91997252cf8e5ddd361737 (diff) | |
download | spark-a680562a6f87a03a00f71bad1c424267ae75c641.tar.gz spark-a680562a6f87a03a00f71bad1c424267ae75c641.tar.bz2 spark-a680562a6f87a03a00f71bad1c424267ae75c641.zip |
[SPARK-14744][EXAMPLES] Clean up examples packaging, remove outdated examples.
First, make all dependencies in the examples module provided, and explicitly
list a couple of ones that somehow are promoted to compile by maven. This
means that to run streaming examples, the streaming connector package needs
to be provided to run-examples using --packages or --jars, just like regular
apps.
Also, remove a couple of outdated examples. HBase has had Spark bindings for
a while and is even including them in the HBase distribution in the next
version, making the examples obsolete. The same applies to Cassandra, which
seems to have a proper Spark binding library already.
I just tested the build, which passes, and ran SparkPi. The examples jars
directory now has only two jars:
```
$ ls -1 examples/target/scala-2.11/jars/
scopt_2.11-3.3.0.jar
spark-examples_2.11-2.0.0-SNAPSHOT.jar
```
Author: Marcelo Vanzin <vanzin@cloudera.com>
Closes #12544 from vanzin/SPARK-14744.
Diffstat (limited to 'examples/src/main/scala')
5 files changed, 0 insertions, 569 deletions
diff --git a/examples/src/main/scala/org/apache/spark/examples/CassandraCQLTest.scala b/examples/src/main/scala/org/apache/spark/examples/CassandraCQLTest.scala deleted file mode 100644 index ca4eea2356..0000000000 --- a/examples/src/main/scala/org/apache/spark/examples/CassandraCQLTest.scala +++ /dev/null @@ -1,135 +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. - */ - - // scalastyle:off println -package org.apache.spark.examples - -import java.nio.ByteBuffer -import java.util.Collections - -import org.apache.cassandra.hadoop.ConfigHelper -import org.apache.cassandra.hadoop.cql3.CqlConfigHelper -import org.apache.cassandra.hadoop.cql3.CqlOutputFormat -import org.apache.cassandra.hadoop.cql3.CqlPagingInputFormat -import org.apache.cassandra.utils.ByteBufferUtil -import org.apache.hadoop.mapreduce.Job - -import org.apache.spark.{SparkConf, SparkContext} - -/* - Need to create following keyspace and column family in cassandra before running this example - Start CQL shell using ./bin/cqlsh and execute following commands - CREATE KEYSPACE retail WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 1}; - use retail; - CREATE TABLE salecount (prod_id text, sale_count int, PRIMARY KEY (prod_id)); - CREATE TABLE ordercf (user_id text, - time timestamp, - prod_id text, - quantity int, - PRIMARY KEY (user_id, time)); - INSERT INTO ordercf (user_id, - time, - prod_id, - quantity) VALUES ('bob', 1385983646000, 'iphone', 1); - INSERT INTO ordercf (user_id, - time, - prod_id, - quantity) VALUES ('tom', 1385983647000, 'samsung', 4); - INSERT INTO ordercf (user_id, - time, - prod_id, - quantity) VALUES ('dora', 1385983648000, 'nokia', 2); - INSERT INTO ordercf (user_id, - time, - prod_id, - quantity) VALUES ('charlie', 1385983649000, 'iphone', 2); -*/ - -/** - * This example demonstrates how to read and write to cassandra column family created using CQL3 - * using Spark. - * Parameters : <cassandra_node> <cassandra_port> - * Usage: ./bin/spark-submit examples.jar \ - * --class org.apache.spark.examples.CassandraCQLTest localhost 9160 - */ -object CassandraCQLTest { - - def main(args: Array[String]) { - val sparkConf = new SparkConf().setAppName("CQLTestApp") - - val sc = new SparkContext(sparkConf) - val cHost: String = args(0) - val cPort: String = args(1) - val KeySpace = "retail" - val InputColumnFamily = "ordercf" - val OutputColumnFamily = "salecount" - - val job = Job.getInstance() - job.setInputFormatClass(classOf[CqlPagingInputFormat]) - val configuration = job.getConfiguration - ConfigHelper.setInputInitialAddress(job.getConfiguration(), cHost) - ConfigHelper.setInputRpcPort(job.getConfiguration(), cPort) - ConfigHelper.setInputColumnFamily(job.getConfiguration(), KeySpace, InputColumnFamily) - ConfigHelper.setInputPartitioner(job.getConfiguration(), "Murmur3Partitioner") - CqlConfigHelper.setInputCQLPageRowSize(job.getConfiguration(), "3") - - /** CqlConfigHelper.setInputWhereClauses(job.getConfiguration(), "user_id='bob'") */ - - /** An UPDATE writes one or more columns to a record in a Cassandra column family */ - val query = "UPDATE " + KeySpace + "." + OutputColumnFamily + " SET sale_count = ? " - CqlConfigHelper.setOutputCql(job.getConfiguration(), query) - - job.setOutputFormatClass(classOf[CqlOutputFormat]) - ConfigHelper.setOutputColumnFamily(job.getConfiguration(), KeySpace, OutputColumnFamily) - ConfigHelper.setOutputInitialAddress(job.getConfiguration(), cHost) - ConfigHelper.setOutputRpcPort(job.getConfiguration(), cPort) - ConfigHelper.setOutputPartitioner(job.getConfiguration(), "Murmur3Partitioner") - - val casRdd = sc.newAPIHadoopRDD(job.getConfiguration(), - classOf[CqlPagingInputFormat], - classOf[java.util.Map[String, ByteBuffer]], - classOf[java.util.Map[String, ByteBuffer]]) - - println("Count: " + casRdd.count) - val productSaleRDD = casRdd.map { - case (key, value) => - (ByteBufferUtil.string(value.get("prod_id")), ByteBufferUtil.toInt(value.get("quantity"))) - } - val aggregatedRDD = productSaleRDD.reduceByKey(_ + _) - aggregatedRDD.collect().foreach { - case (productId, saleCount) => println(productId + ":" + saleCount) - } - - val casoutputCF = aggregatedRDD.map { - case (productId, saleCount) => - val outKey = Collections.singletonMap("prod_id", ByteBufferUtil.bytes(productId)) - val outVal = Collections.singletonList(ByteBufferUtil.bytes(saleCount)) - (outKey, outVal) - } - - casoutputCF.saveAsNewAPIHadoopFile( - KeySpace, - classOf[java.util.Map[String, ByteBuffer]], - classOf[java.util.List[ByteBuffer]], - classOf[CqlOutputFormat], - job.getConfiguration() - ) - - sc.stop() - } -} -// scalastyle:on println diff --git a/examples/src/main/scala/org/apache/spark/examples/CassandraTest.scala b/examples/src/main/scala/org/apache/spark/examples/CassandraTest.scala deleted file mode 100644 index eff840d36e..0000000000 --- a/examples/src/main/scala/org/apache/spark/examples/CassandraTest.scala +++ /dev/null @@ -1,215 +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. - */ - -// scalastyle:off println -package org.apache.spark.examples - -import java.nio.ByteBuffer -import java.util.Arrays -import java.util.SortedMap - -import org.apache.cassandra.db.IColumn -import org.apache.cassandra.hadoop.ColumnFamilyInputFormat -import org.apache.cassandra.hadoop.ColumnFamilyOutputFormat -import org.apache.cassandra.hadoop.ConfigHelper -import org.apache.cassandra.thrift._ -import org.apache.cassandra.utils.ByteBufferUtil -import org.apache.hadoop.mapreduce.Job - -import org.apache.spark.{SparkConf, SparkContext} - -/* - * This example demonstrates using Spark with Cassandra with the New Hadoop API and Cassandra - * support for Hadoop. - * - * To run this example, run this file with the following command params - - * <cassandra_node> <cassandra_port> - * - * So if you want to run this on localhost this will be, - * localhost 9160 - * - * The example makes some assumptions: - * 1. You have already created a keyspace called casDemo and it has a column family named Words - * 2. There are column family has a column named "para" which has test content. - * - * You can create the content by running the following script at the bottom of this file with - * cassandra-cli. - * - */ -object CassandraTest { - - def main(args: Array[String]) { - val sparkConf = new SparkConf().setAppName("casDemo") - // Get a SparkContext - val sc = new SparkContext(sparkConf) - - // Build the job configuration with ConfigHelper provided by Cassandra - val job = Job.getInstance() - job.setInputFormatClass(classOf[ColumnFamilyInputFormat]) - - val host: String = args(1) - val port: String = args(2) - - ConfigHelper.setInputInitialAddress(job.getConfiguration(), host) - ConfigHelper.setInputRpcPort(job.getConfiguration(), port) - ConfigHelper.setOutputInitialAddress(job.getConfiguration(), host) - ConfigHelper.setOutputRpcPort(job.getConfiguration(), port) - ConfigHelper.setInputColumnFamily(job.getConfiguration(), "casDemo", "Words") - ConfigHelper.setOutputColumnFamily(job.getConfiguration(), "casDemo", "WordCount") - - val predicate = new SlicePredicate() - val sliceRange = new SliceRange() - sliceRange.setStart(Array.empty[Byte]) - sliceRange.setFinish(Array.empty[Byte]) - predicate.setSlice_range(sliceRange) - ConfigHelper.setInputSlicePredicate(job.getConfiguration(), predicate) - - ConfigHelper.setInputPartitioner(job.getConfiguration(), "Murmur3Partitioner") - ConfigHelper.setOutputPartitioner(job.getConfiguration(), "Murmur3Partitioner") - - // Make a new Hadoop RDD - val casRdd = sc.newAPIHadoopRDD( - job.getConfiguration(), - classOf[ColumnFamilyInputFormat], - classOf[ByteBuffer], - classOf[SortedMap[ByteBuffer, IColumn]]) - - // Let us first get all the paragraphs from the retrieved rows - val paraRdd = casRdd.map { - case (key, value) => - ByteBufferUtil.string(value.get(ByteBufferUtil.bytes("para")).value()) - } - - // Lets get the word count in paras - val counts = paraRdd.flatMap(p => p.split(" ")).map(word => (word, 1)).reduceByKey(_ + _) - - counts.collect().foreach { - case (word, count) => println(word + ":" + count) - } - - counts.map { - case (word, count) => - val colWord = new org.apache.cassandra.thrift.Column() - colWord.setName(ByteBufferUtil.bytes("word")) - colWord.setValue(ByteBufferUtil.bytes(word)) - colWord.setTimestamp(System.currentTimeMillis) - - val colCount = new org.apache.cassandra.thrift.Column() - colCount.setName(ByteBufferUtil.bytes("wcount")) - colCount.setValue(ByteBufferUtil.bytes(count.toLong)) - colCount.setTimestamp(System.currentTimeMillis) - - val outputkey = ByteBufferUtil.bytes(word + "-COUNT-" + System.currentTimeMillis) - - val mutations = Arrays.asList(new Mutation(), new Mutation()) - mutations.get(0).setColumn_or_supercolumn(new ColumnOrSuperColumn()) - mutations.get(0).column_or_supercolumn.setColumn(colWord) - mutations.get(1).setColumn_or_supercolumn(new ColumnOrSuperColumn()) - mutations.get(1).column_or_supercolumn.setColumn(colCount) - (outputkey, mutations) - }.saveAsNewAPIHadoopFile("casDemo", classOf[ByteBuffer], classOf[List[Mutation]], - classOf[ColumnFamilyOutputFormat], job.getConfiguration) - - sc.stop() - } -} -// scalastyle:on println - -/* -create keyspace casDemo; -use casDemo; - -create column family WordCount with comparator = UTF8Type; -update column family WordCount with column_metadata = - [{column_name: word, validation_class: UTF8Type}, - {column_name: wcount, validation_class: LongType}]; - -create column family Words with comparator = UTF8Type; -update column family Words with column_metadata = - [{column_name: book, validation_class: UTF8Type}, - {column_name: para, validation_class: UTF8Type}]; - -assume Words keys as utf8; - -set Words['3musk001']['book'] = 'The Three Musketeers'; -set Words['3musk001']['para'] = 'On the first Monday of the month of April, 1625, the market - town of Meung, in which the author of ROMANCE OF THE ROSE was born, appeared to - be in as perfect a state of revolution as if the Huguenots had just made - a second La Rochelle of it. Many citizens, seeing the women flying - toward the High Street, leaving their children crying at the open doors, - hastened to don the cuirass, and supporting their somewhat uncertain - courage with a musket or a partisan, directed their steps toward the - hostelry of the Jolly Miller, before which was gathered, increasing - every minute, a compact group, vociferous and full of curiosity.'; - -set Words['3musk002']['book'] = 'The Three Musketeers'; -set Words['3musk002']['para'] = 'In those times panics were common, and few days passed without - some city or other registering in its archives an event of this kind. There were - nobles, who made war against each other; there was the king, who made - war against the cardinal; there was Spain, which made war against the - king. Then, in addition to these concealed or public, secret or open - wars, there were robbers, mendicants, Huguenots, wolves, and scoundrels, - who made war upon everybody. The citizens always took up arms readily - against thieves, wolves or scoundrels, often against nobles or - Huguenots, sometimes against the king, but never against cardinal or - Spain. It resulted, then, from this habit that on the said first Monday - of April, 1625, the citizens, on hearing the clamor, and seeing neither - the red-and-yellow standard nor the livery of the Duc de Richelieu, - rushed toward the hostel of the Jolly Miller. When arrived there, the - cause of the hubbub was apparent to all'; - -set Words['3musk003']['book'] = 'The Three Musketeers'; -set Words['3musk003']['para'] = 'You ought, I say, then, to husband the means you have, however - large the sum may be; but you ought also to endeavor to perfect yourself in - the exercises becoming a gentleman. I will write a letter today to the - Director of the Royal Academy, and tomorrow he will admit you without - any expense to yourself. Do not refuse this little service. Our - best-born and richest gentlemen sometimes solicit it without being able - to obtain it. You will learn horsemanship, swordsmanship in all its - branches, and dancing. You will make some desirable acquaintances; and - from time to time you can call upon me, just to tell me how you are - getting on, and to say whether I can be of further service to you.'; - - -set Words['thelostworld001']['book'] = 'The Lost World'; -set Words['thelostworld001']['para'] = 'She sat with that proud, delicate profile of hers outlined - against the red curtain. How beautiful she was! And yet how aloof! We had been - friends, quite good friends; but never could I get beyond the same - comradeship which I might have established with one of my - fellow-reporters upon the Gazette,--perfectly frank, perfectly kindly, - and perfectly unsexual. My instincts are all against a woman being too - frank and at her ease with me. It is no compliment to a man. Where - the real sex feeling begins, timidity and distrust are its companions, - heritage from old wicked days when love and violence went often hand in - hand. The bent head, the averted eye, the faltering voice, the wincing - figure--these, and not the unshrinking gaze and frank reply, are the - true signals of passion. Even in my short life I had learned as much - as that--or had inherited it in that race memory which we call instinct.'; - -set Words['thelostworld002']['book'] = 'The Lost World'; -set Words['thelostworld002']['para'] = 'I always liked McArdle, the crabbed, old, round-backed, - red-headed news editor, and I rather hoped that he liked me. Of course, Beaumont was - the real boss; but he lived in the rarefied atmosphere of some Olympian - height from which he could distinguish nothing smaller than an - international crisis or a split in the Cabinet. Sometimes we saw him - passing in lonely majesty to his inner sanctum, with his eyes staring - vaguely and his mind hovering over the Balkans or the Persian Gulf. He - was above and beyond us. But McArdle was his first lieutenant, and it - was he that we knew. The old man nodded as I entered the room, and he - pushed his spectacles far up on his bald forehead.'; - -*/ diff --git a/examples/src/main/scala/org/apache/spark/examples/HBaseTest.scala b/examples/src/main/scala/org/apache/spark/examples/HBaseTest.scala deleted file mode 100644 index 65d7489586..0000000000 --- a/examples/src/main/scala/org/apache/spark/examples/HBaseTest.scala +++ /dev/null @@ -1,65 +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. - */ - -// scalastyle:off println -package org.apache.spark.examples - -import org.apache.hadoop.hbase.{HBaseConfiguration, HTableDescriptor, TableName} -import org.apache.hadoop.hbase.client.HBaseAdmin -import org.apache.hadoop.hbase.mapreduce.TableInputFormat - -import org.apache.spark._ - -object HBaseTest { - def main(args: Array[String]) { - val sparkConf = new SparkConf().setAppName("HBaseTest") - val sc = new SparkContext(sparkConf) - - // please ensure HBASE_CONF_DIR is on classpath of spark driver - // e.g: set it through spark.driver.extraClassPath property - // in spark-defaults.conf or through --driver-class-path - // command line option of spark-submit - - val conf = HBaseConfiguration.create() - - if (args.length < 1) { - System.err.println("Usage: HBaseTest <table_name>") - System.exit(1) - } - - // Other options for configuring scan behavior are available. More information available at - // http://hbase.apache.org/apidocs/org/apache/hadoop/hbase/mapreduce/TableInputFormat.html - conf.set(TableInputFormat.INPUT_TABLE, args(0)) - - // Initialize hBase table if necessary - val admin = new HBaseAdmin(conf) - if (!admin.isTableAvailable(args(0))) { - val tableDesc = new HTableDescriptor(TableName.valueOf(args(0))) - admin.createTable(tableDesc) - } - - val hBaseRDD = sc.newAPIHadoopRDD(conf, classOf[TableInputFormat], - classOf[org.apache.hadoop.hbase.io.ImmutableBytesWritable], - classOf[org.apache.hadoop.hbase.client.Result]) - - hBaseRDD.count() - - sc.stop() - admin.close() - } -} -// scalastyle:on println diff --git a/examples/src/main/scala/org/apache/spark/examples/pythonconverters/CassandraConverters.scala b/examples/src/main/scala/org/apache/spark/examples/pythonconverters/CassandraConverters.scala deleted file mode 100644 index 00ce47af48..0000000000 --- a/examples/src/main/scala/org/apache/spark/examples/pythonconverters/CassandraConverters.scala +++ /dev/null @@ -1,70 +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.spark.examples.pythonconverters - -import java.nio.ByteBuffer - -import scala.collection.JavaConverters._ - -import org.apache.cassandra.utils.ByteBufferUtil - -import org.apache.spark.api.python.Converter - -/** - * Implementation of [[org.apache.spark.api.python.Converter]] that converts Cassandra - * output to a Map[String, Int] - */ -class CassandraCQLKeyConverter extends Converter[Any, java.util.Map[String, Int]] { - override def convert(obj: Any): java.util.Map[String, Int] = { - val result = obj.asInstanceOf[java.util.Map[String, ByteBuffer]] - result.asScala.mapValues(ByteBufferUtil.toInt).asJava - } -} - -/** - * Implementation of [[org.apache.spark.api.python.Converter]] that converts Cassandra - * output to a Map[String, String] - */ -class CassandraCQLValueConverter extends Converter[Any, java.util.Map[String, String]] { - override def convert(obj: Any): java.util.Map[String, String] = { - val result = obj.asInstanceOf[java.util.Map[String, ByteBuffer]] - result.asScala.mapValues(ByteBufferUtil.string).asJava - } -} - -/** - * Implementation of [[org.apache.spark.api.python.Converter]] that converts a - * Map[String, Int] to Cassandra key - */ -class ToCassandraCQLKeyConverter extends Converter[Any, java.util.Map[String, ByteBuffer]] { - override def convert(obj: Any): java.util.Map[String, ByteBuffer] = { - val input = obj.asInstanceOf[java.util.Map[String, Int]] - input.asScala.mapValues(ByteBufferUtil.bytes).asJava - } -} - -/** - * Implementation of [[org.apache.spark.api.python.Converter]] that converts a - * List[String] to Cassandra value - */ -class ToCassandraCQLValueConverter extends Converter[Any, java.util.List[ByteBuffer]] { - override def convert(obj: Any): java.util.List[ByteBuffer] = { - val input = obj.asInstanceOf[java.util.List[String]] - input.asScala.map(ByteBufferUtil.bytes).asJava - } -} diff --git a/examples/src/main/scala/org/apache/spark/examples/pythonconverters/HBaseConverters.scala b/examples/src/main/scala/org/apache/spark/examples/pythonconverters/HBaseConverters.scala deleted file mode 100644 index e252ca882e..0000000000 --- a/examples/src/main/scala/org/apache/spark/examples/pythonconverters/HBaseConverters.scala +++ /dev/null @@ -1,84 +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.spark.examples.pythonconverters - -import scala.collection.JavaConverters._ -import scala.util.parsing.json.JSONObject - -import org.apache.hadoop.hbase.CellUtil -import org.apache.hadoop.hbase.KeyValue.Type -import org.apache.hadoop.hbase.client.{Put, Result} -import org.apache.hadoop.hbase.io.ImmutableBytesWritable -import org.apache.hadoop.hbase.util.Bytes - -import org.apache.spark.api.python.Converter - -/** - * Implementation of [[org.apache.spark.api.python.Converter]] that converts all - * the records in an HBase Result to a String - */ -class HBaseResultToStringConverter extends Converter[Any, String] { - override def convert(obj: Any): String = { - val result = obj.asInstanceOf[Result] - val output = result.listCells.asScala.map(cell => - Map( - "row" -> Bytes.toStringBinary(CellUtil.cloneRow(cell)), - "columnFamily" -> Bytes.toStringBinary(CellUtil.cloneFamily(cell)), - "qualifier" -> Bytes.toStringBinary(CellUtil.cloneQualifier(cell)), - "timestamp" -> cell.getTimestamp.toString, - "type" -> Type.codeToType(cell.getTypeByte).toString, - "value" -> Bytes.toStringBinary(CellUtil.cloneValue(cell)) - ) - ) - output.map(JSONObject(_).toString()).mkString("\n") - } -} - -/** - * Implementation of [[org.apache.spark.api.python.Converter]] that converts an - * ImmutableBytesWritable to a String - */ -class ImmutableBytesWritableToStringConverter extends Converter[Any, String] { - override def convert(obj: Any): String = { - val key = obj.asInstanceOf[ImmutableBytesWritable] - Bytes.toStringBinary(key.get()) - } -} - -/** - * Implementation of [[org.apache.spark.api.python.Converter]] that converts a - * String to an ImmutableBytesWritable - */ -class StringToImmutableBytesWritableConverter extends Converter[Any, ImmutableBytesWritable] { - override def convert(obj: Any): ImmutableBytesWritable = { - val bytes = Bytes.toBytes(obj.asInstanceOf[String]) - new ImmutableBytesWritable(bytes) - } -} - -/** - * Implementation of [[org.apache.spark.api.python.Converter]] that converts a - * list of Strings to HBase Put - */ -class StringListToPutConverter extends Converter[Any, Put] { - override def convert(obj: Any): Put = { - val output = obj.asInstanceOf[java.util.ArrayList[String]].asScala.map(Bytes.toBytes).toArray - val put = new Put(output(0)) - put.add(output(1), output(2), output(3)) - } -} |