搜尋
首頁資料庫mysql教程Hadoop 中利用 地图reduce 读写 mysql 数据

Hadoop 中利用 地图reduce 读写 mysql 数据

WBOYWBOYWBOYWBOYWBOYWBOYWBOYWBOYWBOYWBOYWBOYWBOYWB
WBOYWBOYWBOYWBOYWBOYWBOYWBOYWBOYWBOYWBOYWBOYWBOYWB
Jun 07, 2016 pm 04:27 PM
hadoopmysqlreduce使用地圖數位讀寫

Hadoop 中利用 mapreduce 读写 mysql 数据 问题导读 1.hadoop mapreduce的通过哪两个类可以读取数据源？ 2.如果没有mysql驱动包，一般会是什么问题？ 3.如何添加包？ 有时候我们在项目中会遇到输入结果集很大，但是输出结果很小，比如一些 pv、uv 数据，然后

Hadoop 中利用 mapreduce 读写 mysql 数据

问题导读
1.hadoop mapreduce的通过哪两个类可以读取数据源？
2.如果没有mysql驱动包，一般会是什么问题？
3.如何添加包？




有时候我们在项目中会遇到输入结果集很大，但是输出结果很小，比如一些 pv、uv 数据，然后为了实时查询的需求，或者一些 OLAP 的需求，我们需要 mapreduce 与 mysql 进行数据的交互，而这些特性正是 hbase 或者 hive 目前亟待改进的地方。

好了言归正传，简单的说说背景、原理以及需要注意的地方：

1、为了方便 MapReduce 直接访问关系型数据库（Mysql,Oracle），Hadoop提供了DBInputFormat和DBOutputFormat两个类。通过DBInputFormat类把数据库表数据读入到HDFS，根据DBOutputFormat类把MapReduce产生的结果集导入到数据库表中。

2、由于0.20版本对DBInputFormat和DBOutputFormat支持不是很好，该例用了0.19版本来说明这两个类的用法。

至少在我的 0.20.203 中的 org.apache.hadoop.mapreduce.lib 下是没见到 db 包，所以本文也是以老版的 API 来为例说明的。

3、运行MapReduce时候报错：java.io.IOException: com.mysql.jdbc.Driver，一般是由于程序找不到mysql驱动包。解决方法是让每个tasktracker运行MapReduce程序时都可以找到该驱动包。

添加包有两种方式：

（1）在每个节点下的${HADOOP_HOME}/lib下添加该包。重启集群，一般是比较原始的方法。

（2）a)把包传到集群上： hadoop fs -put mysql-connector-java-5.1.0- bin.jar /hdfsPath/

? ?? ? b)在mr程序提交job前，添加语句：DistributedCache.addFileToClassPath(new Path(“/hdfsPath/mysql- connector-java- 5.1.0-bin.jar”), conf);

（3）虽然API用的是0.19的，但是使用0.20的API一样可用，只是会提示方法已过时而已。、

4、测试数据：

  1. CREATE TABLE `t` (
  2. `id` int DEFAULT NULL,
  3. `name` varchar(10) DEFAULT NULL
  4. ) ENGINE=InnoDB DEFAULT CHARSET=utf8;
  6. CREATE TABLE `t2` (
  7. `id` int DEFAULT NULL,
  8. `name` varchar(10) DEFAULT NULL
  9. ) ENGINE=InnoDB DEFAULT CHARSET=utf8;
  11. insert into t values (1,"june"),(2,"decli"),(3,"hello"),
  12. ? ? ? ? (4,"june"),(5,"decli"),(6,"hello"),(7,"june"),
  13. ? ? ? ? (8,"decli"),(9,"hello"),(10,"june"),
  14. ? ? ? ? (11,"june"),(12,"decli"),(13,"hello");
复制代码


5、代码：

  1. import java.io.DataInput;
  2. import java.io.DataOutput;
  3. import java.io.IOException;
  4. import java.sql.PreparedStatement;
  5. import java.sql.ResultSet;
  6. import java.sql.SQLException;
  7. import java.util.Iterator;
  9. import org.apache.hadoop.filecache.DistributedCache;
  10. import org.apache.hadoop.fs.Path;
  11. import org.apache.hadoop.io.LongWritable;
  12. import org.apache.hadoop.io.Text;
  13. import org.apache.hadoop.io.Writable;
  14. import org.apache.hadoop.mapred.JobClient;
  15. import org.apache.hadoop.mapred.JobConf;
  16. import org.apache.hadoop.mapred.MapReduceBase;
  17. import org.apache.hadoop.mapred.Mapper;
  18. import org.apache.hadoop.mapred.OutputCollector;
  19. import org.apache.hadoop.mapred.Reducer;
  20. import org.apache.hadoop.mapred.Reporter;
  21. import org.apache.hadoop.mapred.lib.IdentityReducer;
  22. import org.apache.hadoop.mapred.lib.db.DBConfiguration;
  23. import org.apache.hadoop.mapred.lib.db.DBInputFormat;
  24. import org.apache.hadoop.mapred.lib.db.DBOutputFormat;
  25. import org.apache.hadoop.mapred.lib.db.DBWritable;
  27. /**
  28. * Function: 测试 mr 与 mysql 的数据交互，此测试用例将一个表中的数据复制到另一张表中
  29. * ? ? ? ? ? ? ? ? ? ? ? ???实际当中，可能只需要从 mysql 读，或者写到 mysql 中。
  30. * date: 2013-7-29 上午2:34:04
  31. * @author june
  32. */
  33. public class Mysql2Mr {
  34. ? ? ? ? // DROP TABLE IF EXISTS `hadoop`.`studentinfo`;
  35. ? ? ? ? // CREATE TABLE studentinfo (
  36. ? ? ? ? // id INTEGER NOT NULL PRIMARY KEY,
  37. ? ? ? ? // name VARCHAR(32) NOT NULL);
  39. ? ? ? ? public static class StudentinfoRecord implements Writable, DBWritable {
  40. ? ? ? ? ? ? ? ? int id;
  41. ? ? ? ? ? ? ? ? String name;
  43. ? ? ? ? ? ? ? ? public StudentinfoRecord() {
  45. ? ? ? ? ? ? ? ? }
  47. ? ? ? ? ? ? ? ? public void readFields(DataInput in) throws IOException {
  48. ? ? ? ? ? ? ? ? ? ? ? ? this.id = in.readInt();
  49. ? ? ? ? ? ? ? ? ? ? ? ? this.name = Text.readString(in);
  50. ? ? ? ? ? ? ? ? }
  52. ? ? ? ? ? ? ? ? public String toString() {
  53. ? ? ? ? ? ? ? ? ? ? ? ? return new String(this.id + " " + this.name);
  54. ? ? ? ? ? ? ? ? }
  56. ? ? ? ? ? ? ? ? @Override
  57. ? ? ? ? ? ? ? ? public void write(PreparedStatement stmt) throws SQLException {
  58. ? ? ? ? ? ? ? ? ? ? ? ? stmt.setInt(1, this.id);
  59. ? ? ? ? ? ? ? ? ? ? ? ? stmt.setString(2, this.name);
  60. ? ? ? ? ? ? ? ? }
  62. ? ? ? ? ? ? ? ? @Override
  63. ? ? ? ? ? ? ? ? public void readFields(ResultSet result) throws SQLException {
  64. ? ? ? ? ? ? ? ? ? ? ? ? this.id = result.getInt(1);
  65. ? ? ? ? ? ? ? ? ? ? ? ? this.name = result.getString(2);
  66. ? ? ? ? ? ? ? ? }
  68. ? ? ? ? ? ? ? ? @Override
  69. ? ? ? ? ? ? ? ? public void write(DataOutput out) throws IOException {
  70. ? ? ? ? ? ? ? ? ? ? ? ? out.writeInt(this.id);
  71. ? ? ? ? ? ? ? ? ? ? ? ? Text.writeString(out, this.name);
  72. ? ? ? ? ? ? ? ? }
  73. ? ? ? ? }
  75. ? ? ? ? // 记住此处是静态内部类，要不然你自己实现无参构造器，或者等着抛异常：
  76. ? ? ? ? // Caused by: java.lang.NoSuchMethodException: DBInputMapper.()
  77. ? ? ? ? // http://stackoverflow.com/questions/7154125/custom-mapreduce-input-format-cant-find-constructor
  78. ? ? ? ? // 网上脑残式的转帖，没见到一个写对的。。。
  79. ? ? ? ? public static class DBInputMapper extends MapReduceBase implements
  80. ? ? ? ? ? ? ? ? ? ? ? ? Mapper {
  81. ? ? ? ? ? ? ? ? public void map(LongWritable key, StudentinfoRecord value,
  82. ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? OutputCollector collector, Reporter reporter) throws IOException {
  83. ? ? ? ? ? ? ? ? ? ? ? ? collector.collect(new LongWritable(value.id), new Text(value.toString()));
  84. ? ? ? ? ? ? ? ? }
  85. ? ? ? ? }
  87. ? ? ? ? public static class MyReducer extends MapReduceBase implements
  88. ? ? ? ? ? ? ? ? ? ? ? ? Reducer {
  89. ? ? ? ? ? ? ? ? @Override
  90. ? ? ? ? ? ? ? ? public void reduce(LongWritable key, Iterator values,
  91. ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? OutputCollector output, Reporter reporter) throws IOException {
  92. ? ? ? ? ? ? ? ? ? ? ? ? String[] splits = values.next().toString().split(" ");
  93. ? ? ? ? ? ? ? ? ? ? ? ? StudentinfoRecord r = new StudentinfoRecord();
  94. ? ? ? ? ? ? ? ? ? ? ? ? r.id = Integer.parseInt(splits[0]);
  95. ? ? ? ? ? ? ? ? ? ? ? ? r.name = splits[1];
  96. ? ? ? ? ? ? ? ? ? ? ? ? output.collect(r, new Text(r.name));
  97. ? ? ? ? ? ? ? ? }
  98. ? ? ? ? }
  100. ? ? ? ? public static void main(String[] args) throws IOException {
  101. ? ? ? ? ? ? ? ? JobConf conf = new JobConf(Mysql2Mr.class);
  102. ? ? ? ? ? ? ? ? DistributedCache.addFileToClassPath(new Path("/tmp/mysql-connector-java-5.0.8-bin.jar"), conf);
  104. ? ? ? ? ? ? ? ? conf.setMapOutputKeyClass(LongWritable.class);
  105. ? ? ? ? ? ? ? ? conf.setMapOutputValueClass(Text.class);
  106. ? ? ? ? ? ? ? ? conf.setOutputKeyClass(LongWritable.class);
  107. ? ? ? ? ? ? ? ? conf.setOutputValueClass(Text.class);
  109. ? ? ? ? ? ? ? ? conf.setOutputFormat(DBOutputFormat.class);
  110. ? ? ? ? ? ? ? ? conf.setInputFormat(DBInputFormat.class);
  111. ? ? ? ? ? ? ? ? // // mysql to hdfs
  112. ? ? ? ? ? ? ? ? // conf.setReducerClass(IdentityReducer.class);
  113. ? ? ? ? ? ? ? ? // Path outPath = new Path("/tmp/1");
  114. ? ? ? ? ? ? ? ? // FileSystem.get(conf).delete(outPath, true);
  115. ? ? ? ? ? ? ? ? // FileOutputFormat.setOutputPath(conf, outPath);
  117. ? ? ? ? ? ? ? ? DBConfiguration.configureDB(conf, "com.mysql.jdbc.Driver", "jdbc:mysql://192.168.1.101:3306/test",
  118. ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? "root", "root");
  119. ? ? ? ? ? ? ? ? String[] fields = { "id", "name" };
  120. ? ? ? ? ? ? ? ? // 从 t 表读数据
  121. ? ? ? ? ? ? ? ? DBInputFormat.setInput(conf, StudentinfoRecord.class, "t", null, "id", fields);
  122. ? ? ? ? ? ? ? ? // mapreduce 将数据输出到 t2 表
  123. ? ? ? ? ? ? ? ? DBOutputFormat.setOutput(conf, "t2", "id", "name");
  124. ? ? ? ? ? ? ? ? // conf.setMapperClass(org.apache.hadoop.mapred.lib.IdentityMapper.class);
  125. ? ? ? ? ? ? ? ? conf.setMapperClass(DBInputMapper.class);
  126. ? ? ? ? ? ? ? ? conf.setReducerClass(MyReducer.class);
  128. ? ? ? ? ? ? ? ? JobClient.runJob(conf);
  129. ? ? ? ? }
  130. }
复制代码



6、结果：

执行两次后，你可以看到mysql结果：

  1. mysql> select * from t2;
  2. +------+-------+
  3. | id? ?| name??|
  4. +------+-------+
  5. |? ? 1 | june??|
  6. |? ? 2 | decli |
  7. |? ? 3 | hello |
  8. |? ? 4 | june??|
  9. |? ? 5 | decli |
  10. |? ? 6 | hello |
  11. |? ? 7 | june??|
  12. |? ? 8 | decli |
  13. |? ? 9 | hello |
  14. |? ?10 | june??|
  15. |? ?11 | june??|
  16. |? ?12 | decli |
  17. |? ?13 | hello |
  18. |? ? 1 | june??|
  19. |? ? 2 | decli |
  20. |? ? 3 | hello |
  21. |? ? 4 | june??|
  22. |? ? 5 | decli |
  23. |? ? 6 | hello |
  24. |? ? 7 | june??|
  25. |? ? 8 | decli |
  26. |? ? 9 | hello |
  27. |? ?10 | june??|
  28. |? ?11 | june??|
  29. |? ?12 | decli |
  30. |? ?13 | hello |
  31. +------+-------+
  32. 26 rows in set (0.00 sec)
  34. mysql>
复制代码


7、日志：

  1. 13/07/29 02:33:03 WARN mapred.JobClient: Use GenericOptionsParser for parsing the arguments. Applications should implement Tool for the same.
  2. 13/07/29 02:33:03 INFO filecache.TrackerDistributedCacheManager: Creating mysql-connector-java-5.0.8-bin.jar in /tmp/hadoop-june/mapred/local/archive/-8943686319031389138_-1232673160_640840668/192.168.1.101/tmp-work--8372797484204470322 with rwxr-xr-x
  3. 13/07/29 02:33:03 INFO filecache.TrackerDistributedCacheManager: Cached hdfs://192.168.1.101:9000/tmp/mysql-connector-java-5.0.8-bin.jar as /tmp/hadoop-june/mapred/local/archive/-8943686319031389138_-1232673160_640840668/192.168.1.101/tmp/mysql-connector-java-5.0.8-bin.jar
  4. 13/07/29 02:33:03 INFO filecache.TrackerDistributedCacheManager: Cached hdfs://192.168.1.101:9000/tmp/mysql-connector-java-5.0.8-bin.jar as /tmp/hadoop-june/mapred/local/archive/-8943686319031389138_-1232673160_640840668/192.168.1.101/tmp/mysql-connector-java-5.0.8-bin.jar
  5. 13/07/29 02:33:03 INFO mapred.JobClient: Running job: job_local_0001
  6. 13/07/29 02:33:03 INFO mapred.MapTask: numReduceTasks: 1
  7. 13/07/29 02:33:03 INFO mapred.MapTask: io.sort.mb = 100
  8. 13/07/29 02:33:03 INFO mapred.MapTask: data buffer = 79691776/99614720
  9. 13/07/29 02:33:03 INFO mapred.MapTask: record buffer = 262144/327680
  10. 13/07/29 02:33:03 INFO mapred.MapTask: Starting flush of map output
  11. 13/07/29 02:33:03 INFO mapred.MapTask: Finished spill 0
  12. 13/07/29 02:33:03 INFO mapred.Task: Task:attempt_local_0001_m_000000_0 is done. And is in the process of commiting
  13. 13/07/29 02:33:04 INFO mapred.JobClient:??map 0% reduce 0%
  14. 13/07/29 02:33:06 INFO mapred.LocalJobRunner:?
  15. 13/07/29 02:33:06 INFO mapred.Task: Task 'attempt_local_0001_m_000000_0' done.
  16. 13/07/29 02:33:06 INFO mapred.LocalJobRunner:?
  17. 13/07/29 02:33:06 INFO mapred.Merger: Merging 1 sorted segments
  18. 13/07/29 02:33:06 INFO mapred.Merger: Down to the last merge-pass, with 1 segments left of total size: 235 bytes
  19. 13/07/29 02:33:06 INFO mapred.LocalJobRunner:?
  20. 13/07/29 02:33:06 INFO mapred.Task: Task:attempt_local_0001_r_000000_0 is done. And is in the process of commiting
  21. 13/07/29 02:33:07 INFO mapred.JobClient:??map 100% reduce 0%
  22. 13/07/29 02:33:09 INFO mapred.LocalJobRunner: reduce > reduce
  23. 13/07/29 02:33:09 INFO mapred.Task: Task 'attempt_local_0001_r_000000_0' done.
  24. 13/07/29 02:33:09 WARN mapred.FileOutputCommitter: Output path is null in cleanup
  25. 13/07/29 02:33:10 INFO mapred.JobClient:??map 100% reduce 100%
  26. 13/07/29 02:33:10 INFO mapred.JobClient: Job complete: job_local_0001
  27. 13/07/29 02:33:10 INFO mapred.JobClient: Counters: 18
  28. 13/07/29 02:33:10 INFO mapred.JobClient:? ?File Input Format Counters?
  29. 13/07/29 02:33:10 INFO mapred.JobClient:? ???Bytes Read=0
  30. 13/07/29 02:33:10 INFO mapred.JobClient:? ?File Output Format Counters?
  31. 13/07/29 02:33:10 INFO mapred.JobClient:? ???Bytes Written=0
  32. 13/07/29 02:33:10 INFO mapred.JobClient:? ?FileSystemCounters
  33. 13/07/29 02:33:10 INFO mapred.JobClient:? ???FILE_BYTES_READ=1211691
  34. 13/07/29 02:33:10 INFO mapred.JobClient:? ???HDFS_BYTES_READ=1081704
  35. 13/07/29 02:33:10 INFO mapred.JobClient:? ???FILE_BYTES_WRITTEN=2392844
  36. 13/07/29 02:33:10 INFO mapred.JobClient:? ?Map-Reduce Framework
  37. 13/07/29 02:33:10 INFO mapred.JobClient:? ???Map output materialized bytes=239
  38. 13/07/29 02:33:10 INFO mapred.JobClient:? ???Map input records=13
  39. 13/07/29 02:33:10 INFO mapred.JobClient:? ???Reduce shuffle bytes=0
  40. 13/07/29 02:33:10 INFO mapred.JobClient:? ???Spilled Records=26
  41. 13/07/29 02:33:10 INFO mapred.JobClient:? ???Map output bytes=207
  42. 13/07/29 02:33:10 INFO mapred.JobClient:? ???Map input bytes=13
  43. 13/07/29 02:33:10 INFO mapred.JobClient:? ???SPLIT_RAW_BYTES=75
  44. 13/07/29 02:33:10 INFO mapred.JobClient:? ???Combine input records=0
  45. 13/07/29 02:33:10 INFO mapred.JobClient:? ???Reduce input records=13
  46. 13/07/29 02:33:10 INFO mapred.JobClient:? ???Reduce input groups=13
  47. 13/07/29 02:33:10 INFO mapred.JobClient:? ???Combine output records=0
  48. 13/07/29 02:33:10 INFO mapred.JobClient:? ???Reduce output records=13
  49. 13/07/29 02:33:10 INFO mapred.JobClient:? ???Map output records=13
复制代码







MapReduce直接连接Mysql获取数据






Mysql中数据：

  1. mysql> select * from lxw_tbls;
  2. +---------------------+----------------+
  3. | TBL_NAME? ?? ?? ?? ?| TBL_TYPE? ?? ? |
  4. +---------------------+----------------+
  5. | lxw_test_table? ?? ?| EXTERNAL_TABLE |
  6. | lxw_t? ?? ?? ?? ?? ?| MANAGED_TABLE??|
  7. | lxw_t1? ?? ?? ?? ???| MANAGED_TABLE??|
  8. | tt? ?? ?? ?? ?? ?? ?| MANAGED_TABLE??|
  9. | tab_partition? ?? ? | MANAGED_TABLE??|
  10. | lxw_hbase_table_1? ?| MANAGED_TABLE??|
  11. | lxw_hbase_user_info | MANAGED_TABLE??|
  12. | t? ?? ?? ?? ?? ?? ? | EXTERNAL_TABLE |
  13. | lxw_jobid? ?? ?? ???| MANAGED_TABLE??|
  14. +---------------------+----------------+
  15. 9 rows in set (0.01 sec)
  17. mysql> select * from lxw_tbls where TBL_NAME like 'lxw%' order by TBL_NAME;
  18. +---------------------+----------------+
  19. | TBL_NAME? ?? ?? ?? ?| TBL_TYPE? ?? ? |
  20. +---------------------+----------------+
  21. | lxw_hbase_table_1? ?| MANAGED_TABLE??|
  22. | lxw_hbase_user_info | MANAGED_TABLE??|
  23. | lxw_jobid? ?? ?? ???| MANAGED_TABLE??|
  24. | lxw_t? ?? ?? ?? ?? ?| MANAGED_TABLE??|
  25. | lxw_t1? ?? ?? ?? ???| MANAGED_TABLE??|
  26. | lxw_test_table? ?? ?| EXTERNAL_TABLE |
  27. +---------------------+----------------+
  28. 6 rows in set (0.00 sec)
复制代码


MapReduce程序代码，ConnMysql.java:

  1. package com.lxw.study;
  3. import java.io.DataInput;
  4. import java.io.DataOutput;
  5. import java.io.IOException;
  6. import java.net.URI;
  7. import java.sql.PreparedStatement;
  8. import java.sql.ResultSet;
  9. import java.sql.SQLException;
  10. import java.util.Iterator;
  12. import org.apache.hadoop.conf.Configuration;
  13. import org.apache.hadoop.filecache.DistributedCache;
  14. import org.apache.hadoop.fs.FileSystem;
  15. import org.apache.hadoop.fs.Path;
  16. import org.apache.hadoop.io.LongWritable;
  17. import org.apache.hadoop.io.Text;
  18. import org.apache.hadoop.io.Writable;
  19. import org.apache.hadoop.mapreduce.Job;
  20. import org.apache.hadoop.mapreduce.Mapper;
  21. import org.apache.hadoop.mapreduce.Reducer;
  22. import org.apache.hadoop.mapreduce.lib.db.DBConfiguration;
  23. import org.apache.hadoop.mapreduce.lib.db.DBInputFormat;
  24. import org.apache.hadoop.mapreduce.lib.db.DBWritable;
  25. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
  27. public class ConnMysql {
  28. ? ?? ???
  29. ? ?? ???private static Configuration conf = new Configuration();
  30. ? ?? ???
  31. ? ?? ???static {
  32. ? ?? ?? ?? ?? ? conf.addResource(new Path("F:/lxw-hadoop/hdfs-site.xml"));
  33. ? ?? ?? ?? ?? ? conf.addResource(new Path("F:/lxw-hadoop/mapred-site.xml"));
  34. ? ?? ?? ?? ?? ? conf.addResource(new Path("F:/lxw-hadoop/core-site.xml"));
  35. ? ?? ?? ?? ?? ? conf.set("mapred.job.tracker", "10.133.103.21:50021");
  36. ? ?? ???}
  37. ? ?? ???
  38. ? ?? ???public static class TblsRecord implements Writable, DBWritable {
  39. ? ?? ?? ?? ?? ? String tbl_name;
  40. ? ?? ?? ?? ?? ? String tbl_type;
  42. ? ?? ?? ?? ?? ? public TblsRecord() {
  44. ? ?? ?? ?? ?? ? }
  46. ? ?? ?? ?? ?? ? @Override
  47. ? ?? ?? ?? ?? ? public void write(PreparedStatement statement) throws SQLException {
  48. ? ?? ?? ?? ?? ?? ?? ?? ?// TODO Auto-generated method stub
  49. ? ?? ?? ?? ?? ?? ?? ?? ?statement.setString(1, this.tbl_name);
  50. ? ?? ?? ?? ?? ?? ?? ?? ?statement.setString(2, this.tbl_type);
  51. ? ?? ?? ?? ?? ? }
  53. ? ?? ?? ?? ?? ? @Override
  54. ? ?? ?? ?? ?? ? public void readFields(ResultSet resultSet) throws SQLException {
  55. ? ?? ?? ?? ?? ?? ?? ?? ?// TODO Auto-generated method stub
  56. ? ?? ?? ?? ?? ?? ?? ?? ?this.tbl_name = resultSet.getString(1);
  57. ? ?? ?? ?? ?? ?? ?? ?? ?this.tbl_type = resultSet.getString(2);
  58. ? ?? ?? ?? ?? ? }
  60. ? ?? ?? ?? ?? ? @Override
  61. ? ?? ?? ?? ?? ? public void write(DataOutput out) throws IOException {
  62. ? ?? ?? ?? ?? ?? ?? ?? ?// TODO Auto-generated method stub
  63. ? ?? ?? ?? ?? ?? ?? ?? ?Text.writeString(out, this.tbl_name);
  64. ? ?? ?? ?? ?? ?? ?? ?? ?Text.writeString(out, this.tbl_type);
  65. ? ?? ?? ?? ?? ? }
  67. ? ?? ?? ?? ?? ? @Override
  68. ? ?? ?? ?? ?? ? public void readFields(DataInput in) throws IOException {
  69. ? ?? ?? ?? ?? ?? ?? ?? ?// TODO Auto-generated method stub
  70. ? ?? ?? ?? ?? ?? ?? ?? ?this.tbl_name = Text.readString(in);
  71. ? ?? ?? ?? ?? ?? ?? ?? ?this.tbl_type = Text.readString(in);
  72. ? ?? ?? ?? ?? ? }
  74. ? ?? ?? ?? ?? ? public String toString() {
  75. ? ?? ?? ?? ?? ?? ?? ?? ?return new String(this.tbl_name + " " + this.tbl_type);
  76. ? ?? ?? ?? ?? ? }
  78. ? ?? ???}
  80. ? ?? ???public static class ConnMysqlMapper extends Mapper {
  81. ? ?? ?? ?? ?? ? public void map(LongWritable key,TblsRecord values,Context context)?
  82. ? ?? ?? ?? ?? ?? ?? ?? ?? ?? ???throws IOException,InterruptedException {
  83. ? ?? ?? ?? ?? ?? ?? ?? ?context.write(new Text(values.tbl_name), new Text(values.tbl_type));
  84. ? ?? ?? ?? ?? ? }
  85. ? ?? ???}
  86. ? ?? ???
  87. ? ?? ???public static class ConnMysqlReducer extends Reducer {
  88. ? ?? ?? ?? ?? ? public void reduce(Text key,Iterable values,Context context)?
  89. ? ?? ?? ?? ?? ?? ?? ?? ?? ?? ???throws IOException,InterruptedException {
  90. ? ?? ?? ?? ?? ?? ?? ?? ?for(Iterator itr = values.iterator();itr.hasNext();) {
  91. ? ?? ?? ?? ?? ?? ?? ?? ?? ?? ???context.write(key, itr.next());
  92. ? ?? ?? ?? ?? ?? ?? ?? ?}
  93. ? ?? ?? ?? ?? ? }
  94. ? ?? ???}
  95. ? ?? ???
  96. ? ?? ???public static void main(String[] args) throws Exception {
  97. ? ?? ?? ?? ?? ? Path output = new Path("/user/lxw/output/");
  98. ? ?? ?? ?? ?? ??
  99. ? ?? ?? ?? ?? ? FileSystem fs = FileSystem.get(URI.create(output.toString()), conf);
  100. ? ?? ?? ?? ?? ? if (fs.exists(output)) {
  101. ? ?? ?? ?? ?? ?? ?? ?? ?fs.delete(output);
  102. ? ?? ?? ?? ?? ? }
  103. ? ?? ?? ?? ?? ??
  104. ? ?? ?? ?? ?? ? //mysql的jdbc驱动
  105. ? ?? ?? ?? ?? ? DistributedCache.addFileToClassPath(new Path(??
  106. ? ?? ?? ?? ?? ?? ?? ?? ???"hdfs://hd022-test.nh.sdo.com/user/liuxiaowen/mysql-connector-java-5.1.13-bin.jar"), conf);??
  107. ? ?? ?? ?? ?? ??
  108. ? ?? ?? ?? ?? ? DBConfiguration.configureDB(conf, "com.mysql.jdbc.Driver",??
  109. ? ?? ?? ?? ?? ?? ?? ?? ???"jdbc:mysql://10.133.103.22:3306/hive", "hive", "hive");??
  110. ? ?? ?? ?? ?? ??
  111. ? ?? ?? ?? ?? ? Job job = new Job(conf,"test mysql connection");
  112. ? ?? ?? ?? ?? ? job.setJarByClass(ConnMysql.class);
  113. ? ?? ?? ?? ?? ??
  114. ? ?? ?? ?? ?? ? job.setMapperClass(ConnMysqlMapper.class);
  115. ? ?? ?? ?? ?? ? job.setReducerClass(ConnMysqlReducer.class);
  116. ? ?? ?? ?? ?? ??
  117. ? ?? ?? ?? ?? ? job.setOutputKeyClass(Text.class);
  118. ? ?? ?? ?? ?? ? job.setOutputValueClass(Text.class);
  119. ? ?? ?? ?? ?? ??
  120. ? ?? ?? ?? ?? ? job.setInputFormatClass(DBInputFormat.class);
  121. ? ?? ?? ?? ?? ? FileOutputFormat.setOutputPath(job, output);
  122. ? ?? ?? ?? ?? ??
  123. ? ?? ?? ?? ?? ? //列名
  124. ? ?? ?? ?? ?? ? String[] fields = { "TBL_NAME", "TBL_TYPE" };?
  125. ? ?? ?? ?? ?? ? //六个参数分别为：
  126. ? ?? ?? ?? ?? ? //1.Job;2.Class extends DBWritable>
  127. ? ?? ?? ?? ?? ? //3.表名;4.where条件
  128. ? ?? ?? ?? ?? ? //5.order by语句;6.列名
  129. ? ?? ?? ?? ?? ? DBInputFormat.setInput(job, TblsRecord.class,
  130. ? ?? ?? ?? ?? ?? ?? ?"lxw_tbls", "TBL_NAME like 'lxw%'", "TBL_NAME", fields);??
  131. ? ?? ?? ?? ?? ??
  132. ? ?? ?? ?? ?? ? System.exit(job.waitForCompletion(true) ? 0 : 1);
  133. ? ?? ???}
  134. ? ?? ???
  135. }
复制代码


运行结果：

  1. [lxw@hd025-test ~]$ hadoop fs -cat /user/lxw/output/part-r-00000
  2. lxw_hbase_table_1? ?? ? MANAGED_TABLE
  3. lxw_hbase_user_info? ???MANAGED_TABLE
  4. lxw_jobid? ?? ? MANAGED_TABLE
  5. lxw_t? ?MANAGED_TABLE
  6. lxw_t1??MANAGED_TABLE
  7. lxw_test_table??EXTERNAL_TABLE
复制代码

http://www.aboutyun.com/forum.php?highlight=MapReduce+MySQL&mod=viewthread&tid=7405

陳述
本文內容由網友自願投稿，版權歸原作者所有。本站不承擔相應的法律責任。如發現涉嫌抄襲或侵權的內容，請聯絡admin@php.cn
相關文章
MySQL：世界上最受歡迎的數據庫的簡介MySQL：世界上最受歡迎的數據庫的簡介Apr 12, 2025 am 12:18 AM

MySQL是一種開源的關係型數據庫管理系統，主要用於快速、可靠地存儲和檢索數據。其工作原理包括客戶端請求、查詢解析、執行查詢和返回結果。使用示例包括創建表、插入和查詢數據，以及高級功能如JOIN操作。常見錯誤涉及SQL語法、數據類型和權限問題，優化建議包括使用索引、優化查詢和分錶分區。

MySQL的重要性：數據存儲和管理MySQL的重要性：數據存儲和管理Apr 12, 2025 am 12:18 AM

MySQL是一個開源的關係型數據庫管理系統，適用於數據存儲、管理、查詢和安全。 1.它支持多種操作系統，廣泛應用於Web應用等領域。 2.通過客戶端-服務器架構和不同存儲引擎，MySQL高效處理數據。 3.基本用法包括創建數據庫和表，插入、查詢和更新數據。 4.高級用法涉及復雜查詢和存儲過程。 5.常見錯誤可通過EXPLAIN語句調試。 6.性能優化包括合理使用索引和優化查詢語句。

為什麼要使用mysql？利益和優勢為什麼要使用mysql？利益和優勢Apr 12, 2025 am 12:17 AM

選擇MySQL的原因是其性能、可靠性、易用性和社區支持。 1.MySQL提供高效的數據存儲和檢索功能，支持多種數據類型和高級查詢操作。 2.採用客戶端-服務器架構和多種存儲引擎，支持事務和查詢優化。 3.易於使用，支持多種操作系統和編程語言。 4.擁有強大的社區支持，提供豐富的資源和解決方案。

描述InnoDB鎖定機制（共享鎖，獨家鎖，意向鎖，記錄鎖，間隙鎖，下一鍵鎖）。描述InnoDB鎖定機制（共享鎖，獨家鎖，意向鎖，記錄鎖，間隙鎖，下一鍵鎖）。Apr 12, 2025 am 12:16 AM

InnoDB的鎖機制包括共享鎖、排他鎖、意向鎖、記錄鎖、間隙鎖和下一個鍵鎖。 1.共享鎖允許事務讀取數據而不阻止其他事務讀取。 2.排他鎖阻止其他事務讀取和修改數據。 3.意向鎖優化鎖效率。 4.記錄鎖鎖定索引記錄。 5.間隙鎖鎖定索引記錄間隙。 6.下一個鍵鎖是記錄鎖和間隙鎖的組合，確保數據一致性。

MySQL查詢性能差的常見原因是什麼？MySQL查詢性能差的常見原因是什麼？Apr 12, 2025 am 12:11 AM

MySQL查询性能不佳的原因主要包括没有使用索引、查询优化器选择错误的执行计划、表设计不合理、数据量过大和锁竞争。1.没有索引导致查询缓慢，添加索引后可显著提升性能。2.使用EXPLAIN命令可以分析查询计划，找出优化器错误。3.重构表结构和优化JOIN条件可改善表设计问题。4.数据量大时，采用分区和分表策略。5.高并发环境下，优化事务和锁策略可减少锁竞争。

您什麼時候應該使用複合索引與多個單列索引？您什麼時候應該使用複合索引與多個單列索引？Apr 11, 2025 am 12:06 AM

在數據庫優化中，應根據查詢需求選擇索引策略：1.當查詢涉及多個列且條件順序固定時，使用複合索引；2.當查詢涉及多個列但條件順序不固定時，使用多個單列索引。複合索引適用於優化多列查詢，單列索引則適合單列查詢。

如何識別和優化MySQL中的慢速查詢？ （慢查詢日誌，performance_schema）如何識別和優化MySQL中的慢速查詢？ （慢查詢日誌，performance_schema）Apr 10, 2025 am 09:36 AM

要優化MySQL慢查詢，需使用slowquerylog和performance_schema：1.啟用slowquerylog並設置閾值，記錄慢查詢；2.利用performance_schema分析查詢執行細節，找出性能瓶頸並優化。

MySQL和SQL：開發人員的基本技能MySQL和SQL：開發人員的基本技能Apr 10, 2025 am 09:30 AM

MySQL和SQL是開發者必備技能。 1.MySQL是開源的關係型數據庫管理系統，SQL是用於管理和操作數據庫的標準語言。 2.MySQL通過高效的數據存儲和檢索功能支持多種存儲引擎，SQL通過簡單語句完成複雜數據操作。 3.使用示例包括基本查詢和高級查詢，如按條件過濾和排序。 4.常見錯誤包括語法錯誤和性能問題，可通過檢查SQL語句和使用EXPLAIN命令優化。 5.性能優化技巧包括使用索引、避免全表掃描、優化JOIN操作和提升代碼可讀性。

See all articles

熱AI工具

Undresser.AI Undress

Undresser.AI Undress

人工智慧驅動的應用程序，用於創建逼真的裸體照片

AI Clothes Remover

AI Clothes Remover

用於從照片中去除衣服的線上人工智慧工具。

Undress AI Tool

Undress AI Tool

免費脫衣圖片

Clothoff.io

Clothoff.io

AI脫衣器

AI Hentai Generator

AI Hentai Generator

免費產生 AI 無盡。

顯示更多

熱門文章

R.E.P.O.能量晶體解釋及其做什麼（黃色晶體）
3 週前By尊渡假赌尊渡假赌尊渡假赌
R.E.P.O.最佳圖形設置
3 週前By尊渡假赌尊渡假赌尊渡假赌
刺客信條陰影：貝殼謎語解決方案
2 週前ByDDD
R.E.P.O.如果您聽不到任何人，如何修復音頻
3 週前By尊渡假赌尊渡假赌尊渡假赌
WWE 2K25：如何解鎖Myrise中的所有內容
3 週前By尊渡假赌尊渡假赌尊渡假赌
顯示更多

熱工具

SublimeText3 Linux新版

SublimeText3 Linux新版

SublimeText3 Linux最新版

DVWA

DVWA

Damn Vulnerable Web App (DVWA) 是一個PHP/MySQL的Web應用程序，非常容易受到攻擊。它的主要目標是成為安全專業人員在合法環境中測試自己的技能和工具的輔助工具，幫助Web開發人員更好地理解保護網路應用程式的過程，並幫助教師/學生在課堂環境中教授/學習Web應用程式安全性。 DVWA的目標是透過簡單直接的介面練習一些最常見的Web漏洞，難度各不相同。請注意，該軟體中

ZendStudio 13.5.1 Mac

ZendStudio 13.5.1 Mac

強大的PHP整合開發環境

SecLists

SecLists

SecLists是最終安全測試人員的伙伴。它是一個包含各種類型清單的集合，這些清單在安全評估過程中經常使用，而且都在一個地方。 SecLists透過方便地提供安全測試人員可能需要的所有列表，幫助提高安全測試的效率和生產力。清單類型包括使用者名稱、密碼、URL、模糊測試有效載荷、敏感資料模式、Web shell等等。測試人員只需將此儲存庫拉到新的測試機上，他就可以存取所需的每種類型的清單。

SublimeText3漢化版

SublimeText3漢化版

中文版，非常好用

顯示更多

熱門話題

gmail信箱登陸入口在哪裡
7468
15
CakePHP 教程
1376
52
steam的賬戶名稱是什麼格式
77
11
win11激活密鑰永久
48
19
NYT連接提示和答案
19
28
顯示更多