@hhw 2017-11-10 03:42 采纳率: 0%

MapReducer 写入到数据库报错

【 DBUserWritable 类】

package org.neworigin.com.Database;

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

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

public class DBUserWritable implements DBWritable,WritableComparable{

private String name="";
private String sex="";
private int age=0;
private int num=0;
private String department="";
private String tables="";

@Override
public String toString() {
    return "DBUserWritable [name=" + name + ", sex=" + sex + ", age=" + age + ", department=" + department + "]";
}


public DBUserWritable(DBUserWritable d){
    this.name=d.getName();
    this.sex=d.getSex();
    this.age=d.getAge();
    this.num=d.getNum();
    this.department=d.getDepartment();
    this.tables=d.getTables();
}
public String getName() {
    return name;
}

public void setName(String name) {
    this.name = name;
}

public String getSex() {
    return sex;
}

public void setSex(String sex) {
    this.sex = sex;
}

public int getAge() {
    return age;
}

public void setAge(int age) {
    this.age = age;
}

public int getNum() {
    return num;
}

public void setNum(int num) {
    this.num = num;
}

public String getDepartment() {
    return department;
}

public void setDepartment(String department) {
    this.department = department;
}

public String getTables() {
    return tables;
}

public void setTables(String tables) {
    this.tables = tables;
}

public DBUserWritable(String name, String sex, int age, int num, String department, String tables) {
    super();
    this.name = name;
    this.sex = sex;
    this.age = age;
    this.num = num;
    this.department = department;
    this.tables = tables;
}

public DBUserWritable() {
    super();
    // TODO Auto-generated constructor stub
}

public void write(DataOutput out) throws IOException {
    // TODO Auto-generated method stub
    out.writeUTF(name);
    out.writeUTF(sex);
    out.writeInt(age);
out.writeInt(num);
    out.writeUTF(department);
out.writeUTF(tables);
}

public void readFields(DataInput in) throws IOException {
    // TODO Auto-generated method stub
    name = in.readUTF();
    sex=in.readUTF();
    age=in.readInt();
    num=in.readInt();
    department=in.readUTF();
    tables=in.readUTF();
}

public int compareTo(Object o) {
    // TODO Auto-generated method stub
    return 0;
}

public void write(PreparedStatement statement) throws SQLException {
    // TODO Auto-generated method stub
    statement.setString(1, this.getName());
    statement.setString(2, this.getSex());
    statement.setInt(3, this.getAge());
    statement.setString(4, this.getDepartment());
}

public void readFields(ResultSet resultSet) throws SQLException {
    // TODO Auto-generated method stub
     this.name=resultSet.getString(1);
     this.sex=resultSet.getString(2);
     this.age=resultSet.getInt(3);
     this.department=resultSet.getString(4);
}

}

【mapper】

package org.neworigin.com.Database;

import java.io.IOException;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

public class UserDBMapper extends Mapper {
DBUserWritable DBuser= new DBUserWritable();
@Override
protected void map(LongWritable key, Text value, Mapper.Context context)
throws IOException, InterruptedException {
String[] values=value.toString().split(" ");

    if(values.length==4){

     DBuser.setName(values[0]);
     DBuser.setSex(values[1]);
     DBuser.setAge(Integer.parseInt(values[2]));
     DBuser.setNum(Integer.parseInt(values[3]));
     DBuser.setTables("t1");
     System.out.println("mapper---t1---------------"+DBuser);
     context.write(new Text(values[3]),DBuser);

     }


        if(values.length==2){

             DBuser.setNum(Integer.parseInt(values[0]));
             DBuser.setDepartment(values[1]);   
             DBuser.setTables("t2");
             context.write(new Text(values[0]),DBuser);
             //System.out.println("mapper --t2"+"--"+values[0]+"----"+DBuser);
        }

}

}

【reducer 】

package org.neworigin.com.Database;

import java.io.IOException;
import java.util.LinkedList;
import java.util.List;

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

public class UserDBReducer extends Reducer {
// public DBUserWritable db= new DBUserWritable();

@Override
protected void reduce(Text k2, Iterable<DBUserWritable> v2,
        Reducer<Text, DBUserWritable, NullWritable,DBUserWritable>.Context context) throws IOException, InterruptedException {
    String Name="";
 List<DBUserWritable> list=new LinkedList<DBUserWritable>();
 for(DBUserWritable val : v2){
     list.add(new DBUserWritable(val));//new 一个对象 给list

// System.out.println("[table]"+val.getTables()+"----key"+k2+"---"+val);
if(val.getTables().equals("t2")){
Name=val.getDepartment();

     }
 }
 //键是  num  
for(DBUserWritable join : list){
    System.out.println("[table]"+join.getTables()+"----key"+k2+"---"+join);
if(join.getTables().equals("t1")){

    join.setDepartment(Name);
System.out.println("db-----"+join);
    context.write(NullWritable.get(), join);
    }
}

}

}

【app】

package org.neworigin.com.Database;

import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.db.DBConfiguration;
import org.apache.hadoop.mapreduce.lib.db.DBOutputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class UserDBAPP {

public static void main(String[] args) throws Exception, URISyntaxException {
    // TODO Auto-generated method stub
   String INPUT_PATH="file:///E:/BigData_eclipse_database/Database/data/table1";
   String INPUT_PATH1="file:///E:/BigData_eclipse_database/Database/data/table2";

// String OUTPUT_PARH="file:///E:/BigData_eclipse_database/Database/data/output";
Configuration conf = new Configuration();
// FileSystem fs=FileSystem.get(new URI(OUTPUT_PARH),conf);
// if(fs.exists(new Path(OUTPUT_PARH))){
// fs.delete(new Path(OUTPUT_PARH));
// }

   Job job = new Job(conf,"mydb");
   //设置数据库配置

   DBConfiguration.configureDB(conf, "com.mysql.jdbc.Driver", "jdbc:mysql://localhost/hadoop", "root", "123456");


   FileInputFormat.addInputPaths(job,INPUT_PATH);
   FileInputFormat.addInputPaths(job,INPUT_PATH1);

   job.setMapperClass(UserDBMapper.class);
   job.setMapOutputKeyClass(Text.class);
   job.setMapOutputValueClass(DBUserWritable.class);

   job.setReducerClass(UserDBReducer.class);
   job.setOutputKeyClass(NullWritable.class);
   job.setOutputValueClass(DBUserWritable.class);

// FileOutputFormat.setOutputPath(job, new Path(OUTPUT_PARH));
//设置输出路径
DBOutputFormat.setOutput(job,"user_tables", "name","sex","age","department");
job.setOutputFormatClass(DBOutputFormat.class);
boolean re = job.waitForCompletion(true);
System.out.println(re);
}

}

【报错】ps 表链接，写到本地没问题写到数据库就报错；
17/11/10 11:39:11 WARN output.FileOutputCommitter: Output Path is null in cleanupJob()
17/11/10 11:39:11 WARN mapred.LocalJobRunner: job_local1812680657_0001
java.lang.Exception: java.io.IOException
at org.apache.hadoop.mapred.LocalJobRunner$Job.runTasks(LocalJobRunner.java:462)
at org.apache.hadoop.mapred.LocalJobRunner$Job.run(LocalJobRunner.java:529)
Caused by: java.io.IOException
at org.apache.hadoop.mapreduce.lib.db.DBOutputFormat.getRecordWriter(DBOutputFormat.java:185)
at org.apache.hadoop.mapred.ReduceTask$NewTrackingRecordWriter.(ReduceTask.java:541)
at org.apache.hadoop.mapred.ReduceTask.runNewReducer(ReduceTask.java:614)
at org.apache.hadoop.mapred.ReduceTask.run(ReduceTask.java:389)
at org.apache.hadoop.mapred.LocalJobRunner$Job$ReduceTaskRunnable.run(LocalJobRunner.java:319)
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at java.lang.Thread.run(Thread.java:745)
17/11/10 11:39:12 INFO mapreduce.Job: Job job_local1812680657_0001 running in uber mode : false
17/11/10 11:39:12 INFO mapreduce.Job: map 100% reduce 0%
17/11/10 11:39:12 INFO mapreduce.Job: Job job_local1812680657_0001 failed with state FAILED due to: NA
17/11/10 11:39:12 INFO mapreduce.Job: Counters: 35

写回答
好问题 0 提建议
追加酬金
关注问题
分享
邀请回答
编辑收藏删除结题
收藏举报

6条回答默认最新

关注

码龄粉丝数原力等级 --

被采纳

被点赞

采纳率
荣洋 2017-11-10 06:59
关注
会不会少了端口号3306，jdbc:mysql://localhost/hadoop

解决无用
评论打赏
分享
举报

评论

按下Enter换行，Ctrl+Enter发表内容

报告相同问题？

关注问题

Python 使用多线程写入数据库，报错 python 数据库
2017-11-08 11:46

回答 5 已采纳楼上的原因在于，commit是独占式的，因此在某个线程commit的时候，其它线程执行execute，就会出现楼主的错误。因为对这个线程来说，conn是 not available的，因此 1：建议
python向数据库中写入数据报错 mysql python sql 有问必答
2021-06-05 09:56

回答 4 已采纳如有帮助，望采纳。
使用kepserver用odbc client连接mysql，向数据库内写入值，报错 mysql sql 数据库
2022-01-18 04:05

回答 1 已采纳 odbc是基于绑定数据表的方式来操作的，读取当然没问题，写入的时候，它会连主键一起写入。而你给主键设置的值是0对吗，主键冲突了。如果你上位机一定要用odbc的方式去访问数据库，那数据库里就必须配触发器
系统复习大数据知识点
2020-07-21 18:10

Jecky不姓码的博客 大数据转行不易，希望能帮助到你资料下载+XMind：https://pan.baidu.com/s/1Xn7WPNwNUwdwDW6V0ZGyvg 提取码：raog 复制这段内容后打开百度网盘手机App，操作更方便哦 VX：WL1179952622 有问题的可以加我好友，都是...
python 写入excel文件报错 python 职场和发展问答团队
2022-04-05 15:08

回答 2 已采纳亲亲，第二行尝试使用load_workbook方法
GBase 8a 数据库查询报错或卡住，耗时超长如何解决？数据仓库数据库
2022-01-28 11:03

回答 1 已采纳这种报错通常是磁盘物理部分扇区损坏。需要尝试reboot，看能否恢复。维修、更换磁盘，重做raid，文件系统。数据库做节点替换。
python把列表写入csv报错 python
2022-06-06 12:19

回答 2 已采纳 for a in w : o = ao的值不断被覆盖，所以只有一个值
一个小兔子的大数据见解1
2019-02-20 14:00

会武术的科学家的博客刚去公司的时候，做数据的迁移，写sqoop脚本，（注意：这里可能会问到sqoop增量导入数据的方式式，一般会用到append追加的模式）把数据从oracle数据库导入到hive当中（注意： a.这里我们使用是shell脚本的方式...
spark json数据写入mysql数据库问题 json mysql spark
2022-01-25 09:53

回答 1 已采纳问题已解决：参考这个=====>https://blog.csdn.net/weixin_43753599/article/details/122697542?spm=1001.2014.300
MapReduce将本地数据读入数据库报错 java.io.IOException big data java 其他开发语言
2021-03-31 08:37

回答 3 已采纳你要贴在代码框里，大家可以拷贝了运行，不然，纯人肉看，成本太高，大家就不愿意回答你的问题，你可能就要花很长的时间自己研究。
Python往txt文件写入报错 python
2022-03-08 21:32

回答 1 已采纳不妨考虑对字符串中的单引号进行转义 print('s=\'\'\'<li><a href="http://news.sina.com.cn/o/2022-2-23/a75.shtm
大数据-Hadoop初探-01
2018-04-09 13:24

taco詹詹的博客一，大数据： 1.大数据起源于社交网络 QQ微信发展于电商平台淘宝，京东，一号店 2。***大数据的4V特征体积数据量大TB PB Varity 的数据多种多样文本（.log .txt .sql .xml） Veracity...
php mysql写入查询数据库问题 mysql php sql 数据库
2018-11-19 06:28

回答 6 已采纳看你的数据是否可以覆盖，可以的话使用replace into 不可以的话就设置唯一键
大数据实训笔记
2020-09-17 14:52

qq1772819485的博客 大数据实训笔记（一）大数据实训第一天 Linux常用命令： cd / 切换路径 cd … 回到上一层路径 ls 查看当前目录下所有文件和文件夹 ls -a 显示隐藏文件 ls -l 查看详细信息（ll） mkdir 创建文件夹 touch 创建文件...
09.大数据之Hive基本概念
2018-05-31 22:37

小茗同学IT的博客当然，这个低是有条件的，即数据规模较小，当数据规模大到超过数据库的处理能力的时候，Hive 的并行计算显然能体现出优势。 8. 可扩展性。由于 Hive 是建立在 Hadoop 之上的，因此 Hive 的可扩展性是和 Hadoop ...
没有解决我的问题, 去提问

悬赏问题

¥35 平滑拟合曲线该如何生成
¥100 c语言，请帮蒟蒻写一个题的范例作参考
¥15 名为“Product”的列已属于此 DataTable
¥15 安卓adb backup备份应用数据失败
¥15 eclipse运行项目时遇到的问题
¥15 关于#c##的问题：最近需要用CAT工具Trados进行一些开发
¥15 南大pa1 小游戏没有界面，并且报了如下错误，尝试过换显卡驱动，但是好像不行
¥15 自己瞎改改，结果现在又运行不了了
¥15 链式存储应该如何解决
¥15 没有证书，nginx怎么反向代理到只能接受https的公网网站

MapReducer 写入到数据库 报错

【 DBUserWritable 类 】

【mapper】

【reducer 】

【app】

6条回答 默认 最新

悬赏问题

MapReducer 写入到数据库报错

【 DBUserWritable 类】

6条回答默认最新