mapreduce处理结果向输出至mysql（直接插入/更新/追加式更新）

Rainie999

mapreduce处理结果向输出至mysql（直接插入/更新/追加式更新）
mapreducemysqlupdate 

 
Java代码  

• package cn.m15.ipj.job.usergroup;  
  
•   
  
• Mapreduce处理结果向输出至mysql  
  
•   
  
• 1.写入mysql  
  
•     <1>job中输出的配置：  
  
•         DBOutputFormat.setOutput(this, MySQLConstant.MYSQL_FIX_USER,  
  
•         MySQLConstant.MYSQL_FIX_OPEN_FIRST_FIELDS);  
  
•         (DBOutputFormat为hadoop自带API，将输入插入数据库)  
  
•         public final static String  
  
•         MYSQL_FIX_USER =  
  
•         "ipj_fix_user";  
  
•         public final static String[]  
  
•         MYSQL_FIX_OPEN_FIRST_FIELDS = {"app_id","version","imei","first_open","date"};  
  
•            
  
•     <2>reduce中写入：  
  
•         private FixOpenAppFirstRecord record = new FixOpenAppFirstRecord();  
  
•            
  
•         record.setApp_id(Integer.parseInt(app_id));  
  
•         record.setImei(imei);  
  
•         record.setVersion(version);  
  
•         record.setFirst_open(exactDate);  
  
•         record.setDate(date);  
  
•         context.write(record, NULL);  
  
•            
  
•     <3>FixOpenAppFirstRecord中字段的顺序配置（只列出一条）：  
  
•         @Override  
  
•         public void write(PreparedStatement statement) throws SQLException {  
  
•             statement.setInt(1, app_id);  
  
•             statement.setString(2, version);  
  
•             statement.setString(3, imei);  
  
•             statement.setString(4, first_open);  
  
•             statement.setString(5, date);  
  
•         }  
  
•     <4>注意：  
  
•         1.reduce中record的set值的顺序无所谓，可以任意  
  
•         2.job的mysql字段MYSQL_FIX_OPEN_FIRST_FIELDS的顺序一定要和类FixOpenAppFirstRecord中字段的配置顺序一致  
  
• 2.更新mysql（改变值）  
  
•     <1>job中输出的配置：  
  
•         FixDBOutputFormat.setOutput(this, MySQLConstant.MYSQL_FIX_USER,  
  
•         MySQLConstant.MYSQL_FIX_IS_TAOBAO_FIELDS);  
  
•         (FixDBOutputFormat为自定义Format类，用于更新mysql)  
  
•         public final static String  
  
•         MYSQL_FIX_USER =  
  
•         "ipj_fix_user";  
  
•         public final static String[]  
  
•         MYSQL_FIX_IS_WEIBO_FIELDS = {"is_weibo","app_id","version","imei"};  
  
•     <2>reduce中写入：  
  
•         private FixIsMallUserRecord record = new FixIsMallUserRecord();  
  
•         record.setApp_id(Integer.parseInt(app_id));  
  
•         record.setVersion(version);  
  
•         record.setImei(imei);  
  
•         record.setIs_taobao(is_taobao);  
  
•         context.write(record, NULL);  
  
•     <3>FixIsMallUserRecord 中字段的顺序配置（只列出一条）：  
  
•         @Override  
  
•         public void readFields(ResultSet resultSet) throws SQLException {  
  
•             this.is_taobao = resultSet.getInt(1);  
  
•             this.app_id = resultSet.getInt(2);  
  
•             this.version = resultSet.getString(3);  
  
•             this.imei = resultSet.getString(4);  
  
•         }  
  
•     <4>FixDBOutputFormat中关键的拼接sql代码：  
  
•         public String constructQuery(String table, String[] fieldNames) {  
  
•         if (fieldNames == null) {  
  
•             throw new IllegalArgumentException(  
  
•             "Field names may not be null");  
  
•         }  
  
•            
  
•         StringBuilder query = new StringBuilder();  
  
•         query.append("UPDATE ").append(table);  
  
•         if (fieldNames.length > 0 && fieldNames[0] != null  
  
•         && fieldNames[1] != null&& fieldNames[2] != null  
  
•         && fieldNames[3] != null) {  
  
•             query.append(" SET ");  
  
•             query.append(fieldNames[0] + " =?");  
  
•             query.append(" WHERE ");  
  
•             query.append(fieldNames[1] + " =?");  
  
•             query.append(" AND ");  
  
•             query.append(fieldNames[2] + " =?");  
  
•             query.append(" AND ");  
  
•             query.append(fieldNames[3] + " =?");  
  
•             return query.toString();  
  
•             }  
  
•             return null;  
  
•         }  
  
•     <5>注意：  
  
•         1.reduce中record的set值的顺序无所谓，可以任意  
  
•         2.job的mysql字段MYSQL_FIX_IS_WEIBO_FIELDS的顺序一定要和类  
  
•         3.2中的顺序也一定要和FixDBOutputFormat类中的更新顺序一致（第一个参数为要更新的值，第二三四个参数分别为条件参数）  
  
• 3.更新mysql（值的追加）  
  
•     <1>job中输出的配置：  
  
•         FixAppendDBOutputFormat.setOutput(this, MySQLConstant.MYSQL_FIX_USER,  
  
•         MySQLConstant.MYSQL_FIX_MALL_LOGIN_FIELDS);  
  
•         （FixAppendDBOutputFormat自定义Format，用户更新mysql[追加]）  
  
•         public final static String  
  
•         MYSQL_FIX_USER =  
  
•         "ipj_fix_user";  
  
•         public final static String[]  
  
•         MYSQL_FIX_MALL_LOGIN_FIELDS = {"login_taobao_count","app_id","version","imei"};  
  
•     <2>reduce中写入：  
  
•         private FixMallTotalLoginRecord record = new FixMallTotalLoginRecord();  
  
•         record.setApp_id(Integer.parseInt(app_id));  
  
•         record.setVersion(version);  
  
•         record.setImei(imei);  
  
•         record.setLogin_taobao_count(num);  
  
•         context.write(record, NULL);  
  
•     <3>FixIsMallUserRecord 中字段的顺序配置（只列出一条）：  
  
•         @Override  
  
•         public void write(PreparedStatement statement) throws SQLException {  
  
•             statement.setInt(1, login_taobao_count);  
  
•             statement.setInt(2, app_id);  
  
•             statement.setString(3, version);  
  
•             statement.setString(4, imei);  
  
•         }  
  
•     <4>FixAppendDBOutputFormat中关键的拼接sql代码  
  
•         public String constructQuery(String table, String[] fieldNames) {  
  
•             if (fieldNames == null) {  
  
•                 throw new IllegalArgumentException  
  
•                 ("Field names may not be null");  
  
•             }  
  
•             StringBuilder query = new StringBuilder();  
  
•             query.append("UPDATE ").append(table);  
  
•             if ( fieldNames.length > 0 &&  
  
•             fieldNames[0] != null &&  
  
•             fieldNames[1] != null &&  
  
•             fieldNames[2] != null &&  
  
•             fieldNames[3] != null) {  
  
•                 query.append(" SET ");  
  
•                 query.append(fieldNames[0] +  
  
•                 " = "+fieldNames[0]+"+?");  
  
•                 query.append(" WHERE ");  
  
•                 query.append(fieldNames[1] + " =?");  
  
•                 query.append(" AND ");  
  
•                 query.append(fieldNames[2] + " =?");  
  
•                 query.append(" AND ");  
  
•                 query.append(fieldNames[3] + " =?");  
  
•                 return query.toString();  
  
•                 }  
  
•             return null;  
  
•         }  
  
•     <5>注意：  
  
•         1.reduce中record的set值的顺序无所谓，可以任意  
  
•         2.job的mysql字段MYSQL_FIX_MALL_LOGIN_FIELDS的顺序一定要和类  
  
•         3.2中的顺序也一定要和FixAppendDBOutputFormat类中的更新顺序一致（第一个参数为要更新的值[在原有基础上增加]，第二三四个参数分别为条件参数）