地图reduce处理结果向输出至mysql（直接插入/更新/追加式更新）
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类中的更新顺序一致（第一个参数为要更新的值[在原有基础上增加]，第二三四个参数分别为条件参数）