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

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类中的更新顺序一致（第一个参数为要更新的值[在原有基础上增加]，第二三四个参数分别为条件参数）