Mysql Proxy实现 Mysql读写分离

一、介绍

在我们使用mysql-proxy之前，我们必须先要完成mysql的“主从复制”或者叫“读写分离”，先搭建完成主从复制再使用mysql-proxy。

主从复制分为同步复制和异步复制两种，实际复杂架构中大部分都是异步复制。

主从(Master-Slave)复制的基本过程：

a).Slave上面的IO进程连接上Master，并请求从指定日志文件的指定位置之后的日志内容。

b).Master接收到来自Slave的IO进程的请求后，通过负责复制的IO进程根据请求信息读取指定日志位置之后的日志信息，返回给Slave的IO进程。返回信息中除了日志所包含的信息之外，还包括本次返回的信息已经到Master的bin-log文件的名称以及bin-log的位置。

c).Slave的IO进程接收到信息后，将接收到的日志内容依次添加到Slave端的relay-log文件的最末端，并将读取到的Master的bin-log的文件和位置记录到master-info文件中，以便在下次读取的时候能够清楚的告诉Master（我需要从某个bin-log的那个位置开始往后的日志内容，请发给我）。

d).Slave的sql进程检测到relay-log中新增了内容后，会马上解析relay-log的内容成为在Master上真实执行时候的可执行内容，并在自身执行。

二、环境准备

1、三台Linux虚拟机

2、Linux版本为ubuntukylin-16.04-desktop-amd64

3、Mysql 5.7.20

4、Mysql-Proxy 0.8.1

5、lua 5.2

地址：192.168.1.127（mysql-proxy） 192.168.1.128（master） 192.168.1.129（slave）

三、配置主从复制（读写分离）

3.1安装mysql-server、mysql-client

在master和slave上分别执行：

# sudo apt-get install mysql-server mysql-client

在安装过程中会让你指定root用户的密码：

输入密码：123456

确认密码：123456

3.2 修改master、slave服务器

安装vim编辑器

# sudo apt install vim

编辑：mysqld.cnf

vim /etc/mysql/mysql.conf.d/mysqld.cnf

master机器：

[mysqld]

server-id=1 #设置服务器唯一的id

log-bin=mysql-bin #启用二进制日志

binlog-ignore-db=mysql #忽略写入binlog的库

slave机器：

[mysqld]

server-id=2

replicate-do-db=my #只同步my库（这个可以自己建立）

slave-skip-errors=all #忽略因复制出现的所有错误

3.3 重启主从服务器的mysql

/etc/init.d/mysql restart

3.4 在master上建立帐户并授权slave

这里我使用root用户

GRANT REPLICATION SLAVE ON *.* to "root"@"192.168.1.129" IDENTIFIED BY "123456";

3.5 查看主数据库(master)状态

show master status;

3.6配置从数据库(slave)

mysql > change master to

-> master_host='192.168.1.128',

-> master_port=3306,

-> master_user='root',

-> master_password='123456',

-> master_log_file='mysql-bin.000001',

-> master_log_pos=449;

3.7启动slave同步进程并查看状态

mysql> slave start;

mysql> SHOW SLAVE STATUS\G;

如果显示内容中以下两项都为Yes则表明状态正常。

Slave_IO_Running: Yes

Slave_SQL_Running: Yes

四、测试主从复制

4.1在master上登录

# mysql -uroot -p123456

创建数据库my

创建表student

4.2 在slave上登录

进行如下操作

只要可以在slave上看到你在master上操作产生的数据就说明成功。

五、安装mysql-proxy

MySQL Proxy有一项强大功能是实现“读写分离”，基本原理是让主数据库处理写方面事务，让从库处理SELECT查询。

5.1、安装mysql-proxy

实现读写分离是有lua脚本实现的，现在mysql-proxy里面已经集成，无需再安装。

这里修改ubuntu的源。

# sudo vim /etc/apt/sources.list

将里面内容删除，更换为如下内容

#deb cdrom:[Ubuntu-Kylin 16.04 LTS _Xenial Xerus_ - Release amd64 (20160420.1)]/ xenial main multiverse restricted universe

# See http://help.ubuntu.com/community/UpgradeNotes for how to upgrade to

# newer versions of the distribution.

deb http://cn.archive.ubuntu.com/ubuntu/ xenial main restricted

# deb-src http://cn.archive.ubuntu.com/ubuntu/ xenial main restricted

## Major bug fix updates produced after the final release of the

## distribution.

deb http://cn.archive.ubuntu.com/ubuntu/ xenial-updates main restricted

# deb-src http://cn.archive.ubuntu.com/ubuntu/ xenial-updates main restricted

## N.B. software from this repository is ENTIRELY UNSUPPORTED by the Ubuntu

## team, and may not be under a free licence. Please satisfy yourself as to

## your rights to use the software. Also, please note that software in

## universe WILL NOT receive any review or updates from the Ubuntu security

## team.

deb http://cn.archive.ubuntu.com/ubuntu/ xenial universe

#deb cdrom:[Ubuntu-Kylin 16.04 LTS _Xenial Xerus_ - Release amd64 (20160420.1)]/ xenial main multiverse restricted universe

# See http://help.ubuntu.com/community/UpgradeNotes for how to upgrade to

# newer versions of the distribution.

deb http://cn.archive.ubuntu.com/ubuntu/ xenial main restricted

# deb-src http://cn.archive.ubuntu.com/ubuntu/ xenial main restricted

## Major bug fix updates produced after the final release of the

## distribution.

deb http://cn.archive.ubuntu.com/ubuntu/ xenial-updates main restricted

# deb-src http://cn.archive.ubuntu.com/ubuntu/ xenial-updates main restricted

## N.B. software from this repository is ENTIRELY UNSUPPORTED by the Ubuntu

## team, and may not be under a free licence. Please satisfy yourself as to

## your rights to use the software. Also, please note that software in

## universe WILL NOT receive any review or updates from the Ubuntu security

## team.

deb http://cn.archive.ubuntu.com/ubuntu/ xenial universe

# deb-src http://cn.archive.ubuntu.com/ubuntu/ xenial universe

deb http://cn.archive.ubuntu.com/ubuntu/ xenial-updates universe

# deb-src http://cn.archive.ubuntu.com/ubuntu/ xenial-updates universe

## N.B. software from this repository is ENTIRELY UNSUPPORTED by the Ubuntu

## team, and may not be under a free licence. Please satisfy yourself as to

## your rights to use the software. Also, please note that software in

## multiverse WILL NOT receive any review or updates from the Ubuntu

## security team.

deb http://cn.archive.ubuntu.com/ubuntu/ xenial multiverse

# deb-src http://cn.archive.ubuntu.com/ubuntu/ xenial multiverse

deb http://cn.archive.ubuntu.com/ubuntu/ xenial-updates multiverse

# deb-src http://cn.archive.ubuntu.com/ubuntu/ xenial-updates multiverse

## N.B. software from this repository may not have been tested as

## extensively as that contained in the main release, although it includes

## newer versions of some applications which may provide useful features.

## Also, please note that software in backports WILL NOT receive any review

## or updates from the Ubuntu security team.

deb http://cn.archive.ubuntu.com/ubuntu/ xenial-backports main restricted universe multiverse

# deb-src http://cn.archive.ubuntu.com/ubuntu/ xenial-backports main restricted universe multiverse

## Uncomment the following two lines to add software from Canonical's

## 'partner' repository.

## This software is not part of Ubuntu, but is offered by Canonical and the

## respective vendors as a service to Ubuntu users.

# deb http://archive.canonical.com/ubuntu xenial partner

# deb-src http://archive.canonical.com/ubuntu xenial partner

deb http://security.ubuntu.com/ubuntu xenial-security main restricted

# deb-src http://security.ubuntu.com/ubuntu xenial-security main restricted

deb http://security.ubuntu.com/ubuntu xenial-security universe

# deb-src http://security.ubuntu.com/ubuntu xenial-security universe

deb http://security.ubuntu.com/ubuntu xenial-security multiverse

# deb-src http://security.ubuntu.com/ubuntu xenial-security multiverse

5.2创建mysql-proxy

# vim /etc/mysql-proxy.cnf

[mysql-proxy]

admin-username=root #主从mysql共有的用户

admin-password=123456 #用户的密码

proxy-address=192.168.1.127:4000 #mysql-proxy运行ip和端口，不加端口，默认4040

proxy-read-only-backend-addresses=192.168.1.128 #从库地址

proxy-backend-addresses=192.168.1.129 #主库地址

proxy-lua-script=/usr/share/mysql-proxy/rw-splitting.lua #指定读写分离配置文件位置

admin-lua-script=/usr/share/mysql-proxy/admin-sql.lua #指定管理脚本

log-file=/var/log/mysql-proxy/mysql-proxy.log #日志位置

log-level=info #定义log日志级别，由高到低分别有(error|warning|info|message|debug)

daemon=true #以守护进程方式运行

keepalive=true #mysql-proxy崩溃时，尝试重启

别忘了改变权限：

chmod 660 /etc/mysql-porxy.cnf

5.3 修改读写分离配置文件

默认最小4个最大8个以上的客户端连接才会实现读写分离,现在改为1和1

-- connection pool

if not proxy.global.config.rwsplit then

proxy.global.config.rwsplit = {

min_idle_connections = 1,

max_idle_connections = 1,

is_debug = false

}

end

启动mysql-proxy

# mysql-proxy --defaults-file=/etc/mysql-proxy.cnf

根据mysql-proxy.cnf启动，显示如下：一切ok！

netstat -tupln | grep 4000

tcp 0 0 192.168.1.127:4000 0.0.0.0:* LISTEN 5286/mysql-proxy

5.4测试读写分离

写sql。。。。。。。。。。。。。。略

安装eclipse

解压eclipse-jee-mars-2-win32-x86_64.zip

JavaWordcount

解压spark-2.0.0-bin-hadoop2.6.tgz

创建 Java Project-->Spark

将spark-2.0.0-bin-hadoop2.6下的jars里面的jar全部复制到Spark项目下的lib下

Add Build Path

 1 package com.bean.spark.wordcount;
  2 
  3  
  4 
  5 import java.util.Arrays;
  6 
  7 import java.util.Iterator;
  8 
  9  
 10 
 11 import org.apache.spark.SparkConf;
 12 
 13 import org.apache.spark.api.java.JavaPairRDD;
 14 
 15 import org.apache.spark.api.java.JavaRDD;
 16 
 17 import org.apache.spark.api.java.JavaSparkContext;
 18 
 19 import org.apache.spark.api.java.function.FlatMapFunction;
 20 
 21 import org.apache.spark.api.java.function.Function2;
 22 
 23 import org.apache.spark.api.java.function.PairFunction;
 24 
 25 import org.apache.spark.api.java.function.VoidFunction;
 26 
 27  
 28 
 29 import scala.Tuple2;
 30 
 31  
 32 
 33 public class WordCount {
 34 
 35 public static void main(String[] args) {
 36 
 37 //创建SparkConf对象，设置Spark应用程序的配置信息
 38 
 39 SparkConf conf = new SparkConf();
 40 
 41 conf.setMaster("local");
 42 
 43 conf.setAppName("wordcount");
 44 
 45  
 46 
 47 //创建SparkContext对象，Java开发使用JavaSparkContext；Scala开发使用SparkContext
 48 
 49 //SparkContext负责连接Spark集群，创建RDD、累积量和广播量等
 50 
 51 JavaSparkContext sc = new JavaSparkContext(conf);
 52 
 53  
 54 
 55 //sc中提供了textFile方法是SparkContext中定义的,用来读取HDFS上的
 56 
 57 //文本文件、集群中节点的本地文本文件或任何支持Hadoop的文件系统上的文本文件，它的返回值是JavaRDD[String]，是文本文件每一行
 58 
 59 JavaRDD<String> lines = sc.textFile("D:/tools/data/wordcount/wordcount.txt");
 60 
 61 //将每一行文本内容拆分为多个单词
 62 
 63 //lines调用flatMap这个transformation算子（参数类型是FlatMapFunction接口实现类）返回每一行的每个单词
 64 
 65 JavaRDD<String> words = lines.flatMap(new FlatMapFunction<String, String>() {
 66 
 67  
 68 
 69 private static final long serialVersionUID = 1L;
 70 
 71  
 72 
 73 @Override
 74 
 75 public Iterator<String> call(String s) throws Exception {
 76 
 77 // TODO Auto-generated method stub
 78 
 79 return Arrays.asList(s.split(" ")).iterator();
 80 
 81 }
 82 
 83 });
 84 
 85 //将每个单词的初始数量都标记为1个
 86 
 87 //words调用mapToPair这个transformation算子（参数类型是PairFunction接口实现类，
 88 
 89 //PairFunction<String, String, Integer>的三个参数是<输入单词, Tuple2的key, Tuple2的value>），
 90 
 91 //返回一个新的RDD，即JavaPairRDD
 92 
 93 JavaPairRDD<String, Integer> word = words.mapToPair(new PairFunction<String, String, Integer>() {
 94 
 95  
 96 
 97 private static final long serialVersionUID = 1L;
 98 
 99  
100 
101 @Override
102 
103 public Tuple2<String, Integer> call(String s) throws Exception {
104 
105 // TODO Auto-generated method stub
106 
107 return new Tuple2<String, Integer>(s, 1);
108 
109 }
110 
111 });
112 
113 //计算每个相同单词出现的次数
114 
115 //pairs调用reduceByKey这个transformation算子（参数是Function2接口实现类）对每个key的value进行reduce操作，
116 
117 //返回一个JavaPairRDD，这个JavaPairRDD中的每一个Tuple的key是单词、value则是相同单词次数的和
118 
119 JavaPairRDD<String, Integer> counts = word.reduceByKey(new Function2<Integer, Integer, Integer>() {
120 
121  
122 
123 private static final long serialVersionUID = 1L;
124 
125  
126 
127 @Override
128 
129 public Integer call(Integer s1, Integer s2) throws Exception {
130 
131 // TODO Auto-generated method stub
132 
133 return s1 + s2;
134 
135 }
136 
137 });
138 
139 counts.foreach(new VoidFunction<Tuple2<String,Integer>>() {
140 
141  
142 
143 private static final long serialVersionUID = 1L;
144 
145  
146 
147 @Override
148 
149 public void call(Tuple2<String, Integer> wordcount) throws Exception {
150 
151 // TODO Auto-generated method stub
152 
153 System.out.println(wordcount._1+" : "+wordcount._2);
154 
155 }
156 
157 });
158 
159 //将计算结果文件输出到文件系统
160 
161 /*
162 
163  * HDFS
164 
165  * 新版的API
166 
167  * org.apache.hadoop.mapreduce.lib.output.TextOutputFormat
168 
169  * counts.saveAsNewAPIHadoopFile("hdfs://master:9000/data/wordcount/output", Text.class, IntWritable.class, TextOutputFormat.class, new Configuration());
170 
171  * 使用默认TextOutputFile写入到HDFS(注意写入HDFS权限，如无权限则执行：hdfs dfs -chmod -R 777 /data/wordCount/output)
172 
173          * wordCount.saveAsTextFile("hdfs://soy1:9000/data/wordCount/output");
174 
175          *
176 
177  *
178 
179  * */
180 
181 counts.saveAsTextFile("D:/tools/data/wordcount/output");
182 
183  
184 
185  
186 
187 //关闭SparkContext容器，结束本次作业
188 
189 sc.close();
190 
191 }
192

193 }

运行出错

在代码中加入：只要式加在JavaSparkContext初始化之前就可以

System.setProperty("hadoop.home.dir", "D:/tools/spark-2.0.0-bin-hadoop2.6");

将hadoop2.6(x64)工具.zip解压到D:\tools\spark-2.0.0-bin-hadoop2.6\bin目录下

PythonWordcount

eclipse集成python插件

解压pydev.zip将features和plugins中的包复制到eclipse的对应目录

1 #-*- coding:utf-8-*-
 2 
 3  
 4 
 5 from __future__ import print_function
 6 
 7 from operator import add
 8 
 9 import os
10 
11 from pyspark.context import SparkContext
12 
13 '''
14 
15 wordcount
16 
17 '''
18 
19 if __name__ == "__main__":
20 
21     os.environ["HADOOP_HOME"] = "D:/tools/spark-2.0.0-bin-hadoop2.6"
22 
23     sc = SparkContext()
24 
25     lines = sc.textFile("file:///D:/tools/data/wordcount/wordcount.txt").map(lambda r: r[0:])
26 
27     counts = lines.flatMap(lambda x: x.split(' ')) \
28 
29                   .map(lambda x: (x, 1)) \
30 
31                   .reduceByKey(add)
32 
33     output = counts.collect()
34 
35     for (word, count) in output:
36 
37         print("%s: %i" % (word, count))

提交代码到集群上运行

java：

[hadoop@master application]$ spark-submit --master spark://master:7077 --class com.bean.spark.wordcount.WordCount spark.jar

python：

[hadoop@master application]$ spark-submit --master spark://master:7077 wordcount.py

咸鱼有梦

2017年12月29日星期五