不知道~- 2022-12-19 21:06 采纳率: 54.2%
浏览 37
已结题

expected org.apache.hadoop.io.Text, receivedorg.apache.hadoop.io.LongWritable

img


public class MysqllReadMapper extends Mapper<LongWritable,MysqlWritable,LongWritable,Text> {
    @Override
    protected void map(LongWritable key, MysqlWritable value, Context context) throws IOException, InterruptedException {
        context.write(key,new Text(value.toString()));
    }
}
package com.ll.mysqlreddata;


import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.lib.db.DBInputFormat;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.db.DBConfiguration;
import org.apache.hadoop.mapreduce.lib.input.CombineTextInputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;

public class MysqlReadDriver {
    public MysqlReadDriver()  {
    }

    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
         final Configuration  conf=new Configuration();
        DBConfiguration.configureDB(conf,"com.mysql.jdbc.Driver","jdbc:mysql://localhost:3306/test","root","123456");
        final Job job= Job.getInstance(conf,MysqlReadDriver.class.getSimpleName());
        job.setJarByClass(MysqlReadDriver.class);
        job.setMapperClass(MysqllReadMapper.class);
        job.setReducerClass(MysqlReadReduce.class);

        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(NullWritable.class);
        job.setOutputKeyClass(LongWritable.class);
        job.setOutputValueClass(Text.class);

        DBInputFormat.setInput(job,MysqlWritable.class," SELECT goodsType,AVG(goodsprice),AVG(goodsCount) from  goods_table GROUP BY goodsType","select count(goodsType) from goods_table");
        job.setInputFormatClass(DBInputFormat.class);
        Path output=new Path("./18one");
        FileOutputFormat.setOutputPath(job,output);
//        job.setPartitionerClass(GoodsPartitioner.class);
////    //一般来说 1个分区 就有一个ReduceTask执行该分区的数据
//        job.setNumReduceTasks(2);//与预计的分区一致
        boolean flag=job.waitForCompletion(true);
        System.exit(  flag ?   0 : 1  );
    }
//
    /*
    job.setJarByClass(MysqlReadDriver.class);
        job.setMapperClass(MysqllReadMapper.class);
        job.setReducerClass(MysqlReadReduce.class);
        ./mysqlreadoutput
     */




}

package com.ll.goods_data;

import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

public class MysqlReadReduce extends Reducer<LongWritable,Text,LongWritable,Text> {
    @Override
    protected void reduce(LongWritable key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
        super.reduce(key, values, context);
    }
package com.ll.mysqlreddata;

import org.apache.hadoop.io.Writable;
import org.apache.hadoop.mapreduce.lib.db.DBWritable;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;

public class MysqlWritable implements Writable,DBWritable {
    private String goodsType;
    private double avg_Price;
    private double avg_Count;
    private double goodPrice;

    public double getGoodPrice() {
        return goodPrice;
    }

    public void setGoodPrice(double goodPrice) {
        this.goodPrice = goodPrice;
    }

    public String getGoodsType() {
        return goodsType;
    }

    public void setGoodsType(String goodsType) {
        this.goodsType = goodsType;
    }

    public double getAvg_Price() {
        return avg_Price;
    }

    public void setAvg_Price(double avg_Price) {
        this.avg_Price = avg_Price;
    }

    public double getAvg_Count() {
        return avg_Count;
    }

    public void setAvg_Count(double avg_Count) {
        this.avg_Count = avg_Count;
    }

    @Override
    public String toString() {
        return "类型:" + goodsType+
                "商品的平均价格" +avg_Price  + "平均数量 "+avg_Count;
    }

    public MysqlWritable() {

    }

    @Override
    public void write(PreparedStatement preparedStatement) throws SQLException {
        preparedStatement.setString(1,this.goodsType);
        preparedStatement.setDouble(2,this.avg_Price);
        preparedStatement.setDouble(3,this.avg_Count);


    }

    @Override
    public void readFields(ResultSet resultSet) throws SQLException {
                this.avg_Price=resultSet.getDouble(2);//从结果集获取你想要的数据
                this.avg_Count=resultSet.getDouble(3);//从结果集获取你想要的数据
                this.goodsType=resultSet.getString(1);
//                this.last_update=resultSet.getString("last_update");
    }

    @Override
    public void write(DataOutput dataOutput) throws IOException {
        dataOutput.writeUTF(goodsType);
        dataOutput.writeDouble(getAvg_Price());
              dataOutput.writeDouble(getAvg_Count());
    }

    @Override
    public void readFields(DataInput dataInput) throws IOException {
        this.goodsType=dataInput.readUTF();
        this.avg_Price=dataInput.readDouble();
        this.avg_Count=dataInput.readDouble();

    }
}
  • 写回答

1条回答 默认 最新

  • wux_labs 2022-12-19 22:16
    关注

    类型不匹配,需要文本类型,但接收到了数值类型。
    你的main函数中定义了:

    job.setMapperClass(MysqllReadMapper.class);
    job.setMapOutputKeyClass(Text.class);
    

    也就是你需要map输出Text类型。
    但是你的MysqllReadMapper的输出却定义成了LongWritable,两者类型不匹配。

    public class MysqllReadMapper extends Mapper<LongWritable,MysqlWritable,LongWritable,Text> {
        @Override
        protected void map(LongWritable key, MysqlWritable value, Context context) throws IOException, InterruptedException {
            context.write(key,new Text(value.toString()));
        }
    }
    

    要么改main函数,要么改自定义mapper,改成类型一致。
    望采纳,谢谢!

    本回答被题主选为最佳回答 , 对您是否有帮助呢?
    评论 编辑记录

报告相同问题?

问题事件

  • 系统已结题 12月29日
  • 已采纳回答 12月21日
  • 修改了问题 12月19日
  • 修改了问题 12月19日
  • 展开全部

悬赏问题

  • ¥20 基于MSP430f5529的MPU6050驱动,求出欧拉角
  • ¥20 Java-Oj-桌布的计算
  • ¥15 powerbuilder中的datawindow数据整合到新的DataWindow
  • ¥20 有人知道这种图怎么画吗?
  • ¥15 pyqt6如何引用qrc文件加载里面的的资源
  • ¥15 安卓JNI项目使用lua上的问题
  • ¥20 RL+GNN解决人员排班问题时梯度消失
  • ¥60 要数控稳压电源测试数据
  • ¥15 能帮我写下这个编程吗
  • ¥15 ikuai客户端l2tp协议链接报终止15信号和无法将p.p.p6转换为我的l2tp线路