- android - RelativeLayout 背景可绘制重叠内容
- android - 如何链接 cpufeatures lib 以获取 native android 库?
- java - OnItemClickListener 不起作用,但 OnLongItemClickListener 在自定义 ListView 中起作用
- java - Android 文件转字符串
我正在尝试使用 Apache Cassandra 创建 mapreduce 作业。输入日期来自 cassandra,输出也转到 cassandra。
该程序尝试从名为 tweetstore 的表中选择所有数据,然后插入包含用户名的行数。
这是 mapreduce 作业的主要类:
package com.cassandra.hadoop;
import java.io.*;
import java.lang.*;
import java.util.*;
import java.nio.ByteBuffer;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.util.GenericOptionsParser;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.cassandra.hadoop.ColumnFamilyInputFormat;
import org.apache.cassandra.hadoop.ColumnFamilyOutputFormat;
import org.apache.cassandra.hadoop.ConfigHelper;
import org.apache.cassandra.hadoop.cql3.*;
import org.apache.cassandra.thrift.SlicePredicate;
import org.apache.cassandra.thrift.SliceRange;
import org.apache.cassandra.thrift.IndexExpression;
import org.apache.cassandra.thrift.IndexOperator;
import org.apache.cassandra.utils.ByteBufferUtil;
public class App
{
static final String KEYSPACE_NAME = "tweet_cassandra_map_reduce";
static final String INPUT_COLUMN_FAMILY = "tweetstore";
static final String OUTPUT_COLUMN_FAMILY = "tweetcount";
static final String COLUMN_NAME = "user";
public static void main( String[] args ) throws IOException, InterruptedException, ClassNotFoundException
{
Configuration conf = new Configuration();
String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
Job job = new Job(conf, "tweet count");
job.setJarByClass(App.class);
// mapper configuration.
job.setMapperClass(TweetMapper.class);
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(IntWritable.class);
job.setInputFormatClass(ColumnFamilyInputFormat.class);
// Reducer configuration
job.setReducerClass(TweetAggregator.class);
job.setOutputKeyClass(ByteBuffer.class);
job.setOutputValueClass(List.class);
job.setOutputFormatClass(ColumnFamilyOutputFormat.class);
// Cassandra input column family configuration
ConfigHelper.setInputRpcPort(job.getConfiguration(), "9160");
ConfigHelper.setInputInitialAddress(job.getConfiguration(), "localhost");
ConfigHelper.setInputPartitioner(job.getConfiguration(), "Murmur3Partitioner");
ConfigHelper.setInputColumnFamily(job.getConfiguration(), KEYSPACE_NAME, INPUT_COLUMN_FAMILY);
job.setInputFormatClass(ColumnFamilyInputFormat.class);
SlicePredicate slicePredicate = new SlicePredicate();
slicePredicate.setSlice_range(new SliceRange(ByteBufferUtil.EMPTY_BYTE_BUFFER, ByteBufferUtil.EMPTY_BYTE_BUFFER, false, Integer.MAX_VALUE));
// Prepare index expression.
IndexExpression ixpr = new IndexExpression();ixpr.setColumn_name(ByteBufferUtil.bytes(COLUMN_NAME));
ixpr.setOp(IndexOperator.EQ);
ixpr.setValue(ByteBufferUtil.bytes(otherArgs.length > 0 && !StringUtils.isBlank(otherArgs[0])?otherArgs[0]: "mevivs"));
List<IndexExpression> ixpressions = new ArrayList<IndexExpression>();
ixpressions.add(ixpr);
ConfigHelper.setInputRange(job.getConfiguration(), ixpressions);
ConfigHelper.setInputSlicePredicate(job.getConfiguration(), slicePredicate);
// Cassandra output family configuration.
ConfigHelper.setOutputRpcPort(job.getConfiguration(), "9160");
ConfigHelper.setOutputInitialAddress(job.getConfiguration(), "localhost");
ConfigHelper.setOutputPartitioner(job.getConfiguration(), "Murmur3Partitioner");
ConfigHelper.setOutputColumnFamily(job.getConfiguration(), KEYSPACE_NAME, OUTPUT_COLUMN_FAMILY);
job.setOutputFormatClass(ColumnFamilyOutputFormat.class);
job.getConfiguration().set("row_key", "key");
job.waitForCompletion(true);
}
}
映射器代码
package com.cassandra.hadoop;
import java.io.*;
import java.lang.*;
import java.util.*;
import java.nio.ByteBuffer;
import java.util.SortedMap;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.cassandra.db.Column;
import org.apache.cassandra.utils.ByteBufferUtil;
public class TweetMapper extends Mapper<ByteBuffer, SortedMap<ByteBuffer, Column>, Text, IntWritable>
{
static final String COLUMN_NAME = App.COLUMN_NAME;
private final static IntWritable one = new IntWritable(1);
/* (non-Javadoc)
* @see org.apache.hadoop.mapreduce.Mapper#map(KEYIN, VALUEIN, org.apache.hadoop.mapreduce.
Mapper.Context)
*/
public void map(ByteBuffer key, SortedMap<ByteBuffer, Column> columns, Context context) throws IOException, InterruptedException
{
Column column = columns.get(ByteBufferUtil.bytes(COLUMN_NAME));
String value = ByteBufferUtil.string(column.value());
context.write(new Text(value), one);
}
}
reducer 代码:
package com.cassandra.hadoop;
import java.io.IOException;
import java.util.*;
import java.lang.*;
import java.io.*;
import java.nio.ByteBuffer;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.thrift.Column;
import org.apache.cassandra.thrift.Mutation;
import org.apache.cassandra.thrift.ColumnOrSuperColumn;
import org.apache.cassandra.db.marshal.Int32Type;
public class TweetAggregator extends Reducer<Text,IntWritable, Map<String,ByteBuffer>, List<ByteBuffer>>
{
private static Map<String,ByteBuffer> keys = new HashMap<>();
public void reduce(Text word, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException
{
int sum = 0;
for (IntWritable val : values)
sum += val.get();
System.out.println("writing");
keys.put("key", ByteBufferUtil.bytes(word.toString()));
context.write(keys, getBindVariables(word, sum));
}
private List<ByteBuffer> getBindVariables(Text word, int sum)
{
List<ByteBuffer> variables = new ArrayList<ByteBuffer>();
variables.add(Int32Type.instance.decompose(sum));
return variables;
}
}
当我尝试在 reduce 步骤使用 hadoop cammand
执行作业时出现此错误的问题:
15/02/14 16:53:13 WARN hadoop.AbstractColumnFamilyInputFormat: ignoring jobKeyRange specified without start_key
15/02/14 16:53:14 INFO mapred.JobClient: Running job: job_201502141652_0001
15/02/14 16:53:15 INFO mapred.JobClient: map 0% reduce 0%
15/02/14 16:53:20 INFO mapred.JobClient: map 66% reduce 0%
15/02/14 16:53:22 INFO mapred.JobClient: map 100% reduce 0%
15/02/14 16:53:28 INFO mapred.JobClient: map 100% reduce 33%
15/02/14 16:53:30 INFO mapred.JobClient: Task Id : attempt_201502141652_0001_r_000000_0, Status : FAILED
java.lang.ClassCastException: java.util.HashMap cannot be cast to java.nio.ByteBuffer
at org.apache.cassandra.hadoop.ColumnFamilyRecordWriter.write(ColumnFamilyRecordWriter.java:50)
at org.apache.hadoop.mapred.ReduceTask$NewTrackingRecordWriter.write(ReduceTask.java:588)
at org.apache.hadoop.mapreduce.TaskInputOutputContext.write(TaskInputOutputContext.java:80)
at com.cassandra.hadoop.TweetAggregator.reduce(TweetAggregator.java:40)
at com.cassandra.hadoop.TweetAggregator.reduce(TweetAggregator.java:20)
at org.apache.hadoop.mapreduce.Reducer.run(Reducer.java:176)
at org.apache.hadoop.mapred.ReduceTask.runNewReducer(ReduceTask.java:650)
at org.apache.hadoop.mapred.ReduceTask.run(ReduceTask.java:418)
at org.apache.hadoop.mapred.Child$4.run(Child.java:255)
at java.security.AccessController.doPrivileged(Native Method)
at javax.security.auth.Subject.doAs(Subject.java:415)
at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1136)
at org.apache.hadoop.mapred.Child.main(Child.java:249)
attempt_201502141652_0001_r_000000_0: writing
请帮忙!!谢谢
最佳答案
看起来你的 reducer 作业设置将 bytebuffer 作为输出,而不是 map<>。尝试在您的工作设置中更改此设置
job.setOutputKeyClass(ByteBuffer.class);
对此
job.setOutputKeyClass(Map<String,ByteBuffer>.class);
无论如何,job.set.... 中的泛型类型需要对齐映射器和缩减器的泛型类型参数,因此请检查以确保它们对齐。
关于java - Cassandra Hadoop MapReduce : java. lang.ClassCastException : java. util.HashMap 无法转换为 java.nio.ByteBuffer,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/28517730/
我正在编写一个 mapreduce 应用程序,它接受(键,值)格式的输入并只显示与 reducer 输出相同的数据。 这是示例输入: 1500s 1 1960s 1 Aldus 1 在下面
我不明白,我有一个典型的消息源 content.Language 我可以得到它就像 @Autowire protec
我已经为抽屉导航编写了一个运行良好的程序,但最近我尝试为个人资料图片和 TextView 放置一个 ImageView,之后它给了我一个 ClassCastException。 main_activi
这个问题不太可能帮助任何 future 的访问者;它只与一个小的地理区域、一个特定的时间点或一个非常狭窄的情况有关,这些情况并不普遍适用于互联网的全局受众。为了帮助使这个问题更广泛地适用,visit
@Override public void onPause() { super.onPause(); save(notes.itemSelected); } @Override pub
描述 我正在尝试创建一种自定义语言,我想将词法分析器规则与解析器规则分开。此外,我的目标是将词法分析器和解析器规则进一步划分为特定文件(例如,通用词法分析器规则和关键字规则)。 但我似乎无法让它发挥作
我正在尝试使用以下代码为给定的 Runnable 对象创建代理: public class WorkInvocationHandler implements InvocationHandler {
我有两个非常简单的类,一个扩展了另一个: public class LocationType implements Parcelable { protected int locid =
我遇到了 ClassCastException :Cannot cast my.package.classA to my.package.classA.请注意,(规范)类(名称)是相同的。 我知道这应
我有一个代码试图将一个函数包装在另一个执行动态类型检查的函数中: class Base class Foo extends Base class Bar extends Base object Mai
我使用hadoop 0.18.3遇到以下错误 java.lang.ClassCastException:org.apache.hadoop.io.Text无法转换为org.apache.hadoop.
在 org.dozer.BeanFactory.createBean(Object, Class, String) 的实现中我尝试将对象转换为它的类型。如果我部署所有 bundle ,即关闭并启动所有
我有这个代码: package Maven_test.Maven_project_test; public class Test { class A { int i = 10;
我一直在尝试对 Wicket 的 WebSession 进行子类化,以便可以实现基本的身份验证系统。我已遵循 Wicket 引用库中的指南。当我在网页中尝试以下操作时,出现 ClassCastExce
我正在构建一个 kotlin AAR 库,我需要在发布之前对其进行混淆。我有以下结构: package com.example.token interface TokenManager { s
Kotlin 引入了 here 中描述的声明站点差异. 在某些情况下,泛型参数的 out/in 关键字可能会导致 ClassCastException。我的程序如下所示。 fun main(args:
我正在 AnyLogic 中进行基于代理的传染病流行模拟。我的模型中有两种代理类型 - 人员和建筑物。我正在尝试编写一个函数来计算代理类型“人员”在任何给定时间点所具有的传染性接触数量。下面是我的函数
我有一个 EditContact 类。打开后,它会显示哪些复选框已选中或未选中。这是通过我的适配器中的一些代码完成的,它可以正常工作: //This is for EditContact, t
这个问题已经有答案了: 奥 git _a (2 个回答) 已关闭 5 年前。 我正在获取 ClassCastException 。这两个类来自不同的 jar,但是JettyContinuationPr
我想在 Java 中使用一组对,但是当我调用 contains() 来查看它是否已包含特定对时,我总是会得到 ClassCastException >。有没有办法避免这种行为? 它的实例化如下: pr
我是一名优秀的程序员,十分优秀!