针对cassandra使用hadoop mapreduce的示例代码

mec1mxoz  于 2021-06-03  发布在  Hadoop
关注(0)|答案(2)|浏览(361)

我一直在尝试获取一个运行cassandra的mapreduce示例代码，但遇到运行时错误。源代码：

  1. import java.io.IOException;
  2. import java.nio.ByteBuffer;
  3. import java.util.*;
  4. import java.util.Map.Entry;
  6. import org.apache.cassandra.hadoop.cql3.CqlConfigHelper;
  7. import org.apache.cassandra.hadoop.cql3.CqlOutputFormat;
  8. import org.slf4j.Logger;
  9. import org.slf4j.LoggerFactory;
  11. import org.apache.cassandra.hadoop.cql3.CqlPagingInputFormat;
  12. import org.apache.cassandra.hadoop.ConfigHelper;
  13. import org.apache.cassandra.utils.ByteBufferUtil;
  14. import org.apache.hadoop.conf.Configuration;
  15. import org.apache.hadoop.conf.Configured;
  16. import org.apache.hadoop.fs.Path;
  17. import org.apache.hadoop.io.IntWritable;
  18. import org.apache.hadoop.io.Text;
  19. import org.apache.hadoop.mapreduce.Job;
  20. import org.apache.hadoop.mapreduce.Mapper;
  21. import org.apache.hadoop.mapreduce.Reducer;
  22. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
  23. import org.apache.hadoop.util.Tool;
  24. import org.apache.hadoop.util.ToolRunner;
  26. import java.nio.charset.CharacterCodingException;
  28. /**
  29.  * This counts the occurrences of words in ColumnFamily
  30.  *   cql3_worldcount ( user_id text,
  31.  *                   category_id text,
  32.  *                   sub_category_id text,
  33.  *                   title  text,
  34.  *                   body  text,
  35.  *                   PRIMARY KEY (user_id, category_id, sub_category_id))
  36.  *
  37.  * For each word, we output the total number of occurrences across all body texts.
  38.  *
  39.  * When outputting to Cassandra, we write the word counts to column family
  40.  *  output_words ( row_id1 text,
  41.  *                 row_id2 text,
  42.  *                 word text,
  43.  *                 count_num text,
  44.  *                 PRIMARY KEY ((row_id1, row_id2), word))
  45.  * as a {word, count} to columns: word, count_num with a row key of "word sum"
  46.  */
  47. public class WordCount extends Configured implements Tool
  48. {
  49.     private static final Logger logger = LoggerFactory.getLogger(WordCount.class);
  51.     static final String KEYSPACE = "cql3_worldcount";
  52.     static final String COLUMN_FAMILY = "inputs";
  54.     static final String OUTPUT_REDUCER_VAR = "output_reducer";
  55.     static final String OUTPUT_COLUMN_FAMILY = "output_words";
  57.     private static final String OUTPUT_PATH_PREFIX = "/tmp/word_count";
  59.     private static final String PRIMARY_KEY = "row_key";
  61.     public static void main(String[] args) throws Exception
  62.     {
  63.         // Let ToolRunner handle generic command-line options
  64.         ToolRunner.run(new Configuration(), new WordCount(), args);
  65.         System.exit(0);
  66.     }
  68.     public static class TokenizerMapper extends Mapper<Map<String, ByteBuffer>, Map<String, ByteBuffer>, Text, IntWritable>
  69.     {
  70.         private final static IntWritable one = new IntWritable(1);
  71.         private Text word = new Text();
  72.         private ByteBuffer sourceColumn;
  74.         protected void setup(org.apache.hadoop.mapreduce.Mapper.Context context)
  75.         throws IOException, InterruptedException
  76.         {
  77.         }
  79.         public void map(Map<String, ByteBuffer> keys, Map<String, ByteBuffer> columns, Context context) throws IOException, InterruptedException
  80.         {
  81.             for (Entry<String, ByteBuffer> column : columns.entrySet())
  82.             {
  83.                 if (!"body".equalsIgnoreCase(column.getKey()))
  84.                     continue;
  86.                 String value = ByteBufferUtil.string(column.getValue());
  88.                 logger.debug("read {}:{}={} from {}",
  89.                              new Object[] {toString(keys), column.getKey(), value, context.getInputSplit()});
  91.                 StringTokenizer itr = new StringTokenizer(value);
  92.                 while (itr.hasMoreTokens())
  93.                 {
  94.                     word.set(itr.nextToken());
  95.                     context.write(word, one);
  96.                 }
  97.             }
  98.         }
  100.         private String toString(Map<String, ByteBuffer> keys)
  101.         {
  102.             String result = "";
  103.             try
  104.             {
  105.                 for (ByteBuffer key : keys.values())
  106.                     result = result + ByteBufferUtil.string(key) + ":";
  107.             }
  108.             catch (CharacterCodingException e)
  109.             {
  110.                 logger.error("Failed to print keys", e);
  111.             }
  112.             return result;
  113.         }
  114.     }
  116.     public static class ReducerToFilesystem extends Reducer<Text, IntWritable, Text, IntWritable>
  117.     {
  118.         public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException
  119.         {
  120.             int sum = 0;
  121.             for (IntWritable val : values)
  122.                 sum += val.get();
  123.             context.write(key, new IntWritable(sum));
  124.         }
  125.     }
  127.     public static class ReducerToCassandra extends Reducer<Text, IntWritable, Map<String, ByteBuffer>, List<ByteBuffer>>
  128.     {
  129.         private Map<String, ByteBuffer> keys;
  130.         private ByteBuffer key;
  131.         protected void setup(org.apache.hadoop.mapreduce.Reducer.Context context)
  132.         throws IOException, InterruptedException
  133.         {
  134.             keys = new LinkedHashMap<String, ByteBuffer>();
  135.             String[] partitionKeys = context.getConfiguration().get(PRIMARY_KEY).split(",");
  136.             keys.put("row_id1", ByteBufferUtil.bytes(partitionKeys[0]));
  137.             keys.put("row_id2", ByteBufferUtil.bytes(partitionKeys[1]));
  138.         }
  140.         public void reduce(Text word, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException
  141.         {
  142.             int sum = 0;
  143.             for (IntWritable val : values)
  144.                 sum += val.get();
  145.             context.write(keys, getBindVariables(word, sum));
  146.         }
  148.         private List<ByteBuffer> getBindVariables(Text word, int sum)
  149.         {
  150.             List<ByteBuffer> variables = new ArrayList<ByteBuffer>();
  151.             keys.put("word", ByteBufferUtil.bytes(word.toString()));
  152.             variables.add(ByteBufferUtil.bytes(String.valueOf(sum)));         
  153.             return variables;
  154.         }
  155.     }
  157.     public int run(String[] args) throws Exception
  158.     {
  159.         String outputReducerType = "filesystem";
  160.         if (args != null && args[0].startsWith(OUTPUT_REDUCER_VAR))
  161.         {
  162.             String[] s = args[0].split("=");
  163.             if (s != null && s.length == 2)
  164.                 outputReducerType = s[1];
  165.         }
  166.         logger.info("output reducer type: " + outputReducerType);
  168.         Job job = new Job(getConf(), "wordcount");
  169.         job.setJarByClass(WordCount.class);
  170.         job.setMapperClass(TokenizerMapper.class);
  172.         if (outputReducerType.equalsIgnoreCase("filesystem"))
  173.         {
  174.             job.setCombinerClass(ReducerToFilesystem.class);
  175.             job.setReducerClass(ReducerToFilesystem.class);
  176.             job.setOutputKeyClass(Text.class);
  177.             job.setOutputValueClass(IntWritable.class);
  178.             FileOutputFormat.setOutputPath(job, new Path(OUTPUT_PATH_PREFIX));
  179.         }
  180.         else
  181.         {
  182.             job.setReducerClass(ReducerToCassandra.class);
  184.             job.setMapOutputKeyClass(Text.class);
  185.             job.setMapOutputValueClass(IntWritable.class);
  186.             job.setOutputKeyClass(Map.class);
  187.             job.setOutputValueClass(List.class);
  189.             job.setOutputFormatClass(CqlOutputFormat.class);
  191.             ConfigHelper.setOutputColumnFamily(job.getConfiguration(), KEYSPACE, OUTPUT_COLUMN_FAMILY);
  192.             job.getConfiguration().set(PRIMARY_KEY, "word,sum");
  193.             String query = "UPDATE " + KEYSPACE + "." + OUTPUT_COLUMN_FAMILY +
  194.                            " SET count_num = ? ";
  195.             CqlConfigHelper.setOutputCql(job.getConfiguration(), query);
  196.             ConfigHelper.setOutputInitialAddress(job.getConfiguration(), "localhost");
  197.             ConfigHelper.setOutputPartitioner(job.getConfiguration(), "Murmur3Partitioner");
  198.         }
  200.         job.setInputFormatClass(CqlPagingInputFormat.class);
  202.         ConfigHelper.setInputRpcPort(job.getConfiguration(), "9160");
  203.         ConfigHelper.setInputInitialAddress(job.getConfiguration(), "localhost");
  204.         ConfigHelper.setInputColumnFamily(job.getConfiguration(), KEYSPACE, COLUMN_FAMILY);
  205.         ConfigHelper.setInputPartitioner(job.getConfiguration(), "Murmur3Partitioner");
  207.         CqlConfigHelper.setInputCQLPageRowSize(job.getConfiguration(), "3");
  208.         //this is the user defined filter clauses, you can comment it out if you want count all titles
  209.         CqlConfigHelper.setInputWhereClauses(job.getConfiguration(), "title='A'");
  210.         job.waitForCompletion(true);
  211.         return 0;
  212.     }
  213. }

它编译得很好，但是我得到了一个错误：

  1. Exception in thread "main" java.lang.NoClassDefFoundError: org/apache/cassandra/hadoop/cql3/CqlPagingInputFormat
  2.         at WordCount.run(WordCount.java:230)
  3.         at org.apache.hadoop.util.ToolRunner.run(ToolRunner.java:65)
  4.         at WordCount.main(WordCount.java:94)
  5.         at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
  6.         at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
  7.         at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
  8.         at java.lang.reflect.Method.invoke(Method.java:606)
  9.         at org.apache.hadoop.util.RunJar.main(RunJar.java:160)
  10. Caused by: java.lang.ClassNotFoundException: org.apache.cassandra.hadoop.cql3.CqlPagingInputFormat
  11.         at java.net.URLClassLoader$1.run(URLClassLoader.java:366)
  12.         at java.net.URLClassLoader$1.run(URLClassLoader.java:355)
  13.         at java.security.AccessController.doPrivileged(Native Method)
  14.         at java.net.URLClassLoader.findClass(URLClassLoader.java:354)
  15.         at java.lang.ClassLoader.loadClass(ClassLoader.java:425)
  16.         at java.lang.ClassLoader.loadClass(ClassLoader.java:358)
  17.         ... 8 more

我正在使用hadoop1.2.1和cassandra2.0.4。帮助这个错误或样本代码或指令让hadoopmapreduce与cassandra一起工作将不胜感激。

hadoopcassandramapreduce

2条答案

按热度按时间
dy1byipe

dy1byipe1#

请使用以下路径
在//conf/hadoop-env.sh文件中导出hadoop\u classpath=//lib/*：$hadoop\u classpath。

赞(0）分享  回复(0）举报  2021-06-03
6yjfywim

6yjfywim2#

要解决这个问题，请将cassandrajar文件复制到hadoop lib目录。

赞(0）分享  回复(0）举报  2021-06-03
我来回答

相关问题

    查看更多

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新问答

    更多
    • 蜀ICP备13028337号-1 大数据知识库 https://www.saoniuhuo.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com