docker--doublehappy

2020-04-27

docker 一定是选用  社区版本的 花钱的版本我们不用的

docker-linux-centos

部署

在这里插入图片描述

按照官网走即可:

1.部署

安装 yum工具：
yum install -y yum-utils \
device-mapper-persistent-data \
lvm2
----------------------------------------------
安装dockeryum源：

yum-config-manager \
--add-repo \
https://download.docker.com/linux/centos/docker-ce.repo
----------------------------------------------
安装docker的引擎：

yum install -y docker-ce docker-ce-cli containerd.io
----------------------------------------------
启动docker：

systemctl start docker
systemctl status docker

----------------------------------------------

注意：
	上面的第二步骤：
[root@hadoop101 ~]# yum-config-manager \
> --add-repo \
> https://download.docker.com/linux/centos/docker-ce.repo
Loaded plugins: fastestmirror
adding repo from: https://download.docker.com/linux/centos/docker-ce.repo
grabbing file https://download.docker.com/linux/centos/docker-ce.repo to /etc/yum.repos.d/docker-ce.repo
repo saved to /etc/yum.repos.d/docker-ce.repo
[root@hadoop101 ~]# ll /etc/yum.repos.d
total 16
-rw-r--r-- 1 root root  675 Sep 11 09:44 CentOS-Base.repo
-rw-r--r-- 1 root root 2424 Oct 19 05:57 docker-ce.repo
-rw-r--r-- 1 root root  230 Sep 11 09:44 epel.repo
-rw-r--r-- 1 root root  113 Jul 15  2014 nginx.repo
[root@hadoop101 ~]# 

/etc/yum.repos.d：
	这个目录就是你 yum 源所安装的东西 

可以看一下：
[root@hadoop101 ~]# cat /etc/yum.repos.d/docker-ce.repo
[docker-ce-stable]
name=Docker CE Stable - $basearch
baseurl=https://download.docker.com/linux/centos/7/$basearch/stable
enabled=1
gpgcheck=1
gpgkey=https://download.docker.com/linux/centos/gpg

[docker-ce-stable-debuginfo]
name=Docker CE Stable - Debuginfo $basearch
baseurl=https://download.docker.com/linux/centos/7/debug-$basearch/stable
enabled=
gpgcheck=1
gpgkey=https://download.docker.com/linux/centos/gpg
......


这就是 yum源 到时候去这些地址 下载东西

docker hub

docker官方的镜像仓库地址: 与github是类似的 

注意：
1.docker hub 就是存放我们的镜像的
2.image 镜像
3.container 镜像文件创建的实例

eg：
	win7 系统盘 iso镜像文件 ==》VM工具==》win7-1  win7-2 实例
即：
	win7 系统盘 iso镜像文件    就是 image
	win7-1  win7-2 实例       就是 container

docker官方的镜像仓库地址

案例

hello world

docker help 

[root@hadoop101 ~]# docker search hello
NAME                                       DESCRIPTION                                     STARS               OFFICIAL            AUTOMATED
hello-world                                Hello World! (an example of minimal Dockeriz…   1121                [OK]                
kitematic/hello-world-nginx                A light-weight nginx container that demonstr…   138                                     

我们找一个案例  hello-world  官网的

1.先把 需要的 pull 下来 

[root@hadoop101 ~]# docker pull hello-world
Using default tag: latest
latest: Pulling from library/hello-world
1b930d010525: Pull complete 
Digest: sha256:9572f7cdcee8591948c2963463447a53466950b3fc15a247fcad1917ca215a2f
Status: Downloaded newer image for hello-world:latest
docker.io/library/hello-world:latest
[root@hadoop101 ~]# 

	hello-world:latest
指的是：
	镜像的名字 ： 镜像的版本

2.看一下我们机器上有那些镜像

[root@hadoop101 ~]# docker images
REPOSITORY          TAG                 IMAGE ID            CREATED             SIZE
apache/kudu         latest              e1c6e5b62086        2 months ago        476MB
hello-world         latest              fce289e99eb9        13 months ago       1.84kB
kunickiaj/kudu      latest              c30c95f5fe58        17 months ago       896MB
docker/compose      1.8.0               89188432ef03        3 years ago         59.1MB
[root@hadoop101 ~]# 

3. 通过 镜像模版 生成一个 容器（实例）
 	  通过 image 生成一个  container

[root@hadoop101 ~]# docker run hello-world

Hello from Docker!
This message shows that your installation appears to be working correctly.

To generate this message, Docker took the following steps:
 1. The Docker client contacted the Docker daemon.
 2. The Docker daemon pulled the "hello-world" image from the Docker Hub.
    (amd64)
 3. The Docker daemon created a new container from that image which runs the
    executable that produces the output you are currently reading.
 4. The Docker daemon streamed that output to the Docker client, which sent it
    to your terminal.

To try something more ambitious, you can run an Ubuntu container with:
 $ docker run -it ubuntu bash

Share images, automate workflows, and more with a free Docker ID:
 https://hub.docker.com/

For more examples and ideas, visit:
 https://docs.docker.com/get-started/

[root@hadoop101 ~]# 

4.查看运行生成的实例
[root@hadoop101 ~]# docker ps -a
CONTAINER ID        IMAGE                COMMAND                  CREATED              STATUS                          PORTS               NAMES
a38e8df475a1        hello-world          "/hello"                 About a minute ago   Exited (0) About a minute ago                       strange_thompson
cc897cbba1ae        kunickiaj/kudu       "/docker-entrypoint.…"   8 weeks ago          Exited (137) 6 weeks ago                            kudu-tserver
5402fe1c9d3a        kunickiaj/kudu       "/docker-entrypoint.…"   8 weeks ago          Exited (137) 6 weeks ago                            kudu-master
c36ac4effd05        kunickiaj/kudu       "/docker-entrypoint.…"   8 weeks ago          Created                                             kudu-tserver-data
37f0ad272538        kunickiaj/kudu       "/docker-entrypoint.…"   8 weeks ago          Created                                             kudu-master-data
ee07f7ee2717        apache/kudu:latest   "/kudu-entrypoint.sh…"   8 weeks ago          Exited (137) 6 weeks ago                            docker_kudu-tserver_1
1a1ae66602ad        apache/kudu:latest   "/kudu-entrypoint.sh…"   8 weeks ago          Exited (137) 6 weeks ago                            docker_kudu-master-1_1
f440472b35b2        apache/kudu:latest   "/kudu-entrypoint.sh…"   8 weeks ago          Exited (137) 6 weeks ago                            docker_kudu-master-2_1
1b32425575bb        apache/kudu:latest   "/kudu-entrypoint.sh…"   8 weeks ago          Exited (137) 6 weeks ago                            docker_kudu-master-3_1
[root@hadoop101 ~]# 

注意：
第三波的打印：就是 docker的步骤

1.docker 有 客户端 + 服务端 
	即：docker命令 就是 client
		后台 status 显示的进程 就是 daemon（服务段）
2.client 仅仅 用来 发送指令的 
	其他的 run 等 全是 服务段来做的

nginx

在这里插入图片描述

去 docker hub 上面 搜一下 使用官方的

nginx的dock！！！
在这里插入图片描述

在这里插入图片描述

在上面的url 的 描述选项 可以知道如何使用 nginx：

1.docker pull nginx:latest

2.docker run \
--name doublehappy-nginx1 \
-d -p 8080:80 \
nginx:latest

注意：
--name 指定 镜像实例的名字
nginx:latest  指的是 你来自哪个镜像的 
8080:80     
	8080 指linux的端口  80 指的是 容器的端口 即 我访问linux8080 会映射到docker 80 端口
 
3. 查看现在存活的 
[root@hadoop101 ~]# docker ps 
CONTAINER ID        IMAGE               COMMAND                  CREATED             STATUS              PORTS                  NAMES
d4267c461c6c        nginx:latest        "nginx -g 'daemon of…"   40 seconds ago      Up 40 seconds       0.0.0.0:8080->80/tcp   doublehappy-nginx1
[root@hadoop101 ~]# 

加上 -a  连推出过的 都看

注意：
 0.0.0.0:8080->80/tcp  
试着 页面访问 通过 linux 8080

在这里插入图片描述

说明：
	我们通过docker 启动的一个 实例 
	容器里 运行着nginx 是可以访问的

4.进入 nginx 的 
[root@hadoop101 ~]# docker exec -it d4267c461c6c /bin/bash
root@d4267c461c6c:/# 

看一个东西：
root@d4267c461c6c:/# ls -l /
total 64
drwxr-xr-x  2 root root 4096 Jan 30 00:00 bin
drwxr-xr-x  2 root root 4096 Nov 10 12:17 boot
drwxr-xr-x  5 root root  340 Feb  6 07:49 dev
drwxr-xr-x  1 root root 4096 Feb  6 07:49 etc
drwxr-xr-x  2 root root 4096 Nov 10 12:17 home
drwxr-xr-x  1 root root 4096 Feb  2 08:06 lib
drwxr-xr-x  2 root root 4096 Jan 30 00:00 lib64
drwxr-xr-x  2 root root 4096 Jan 30 00:00 media
drwxr-xr-x  2 root root 4096 Jan 30 00:00 mnt
drwxr-xr-x  2 root root 4096 Jan 30 00:00 opt
dr-xr-xr-x 93 root root    0 Feb  6 07:49 proc
drwx------  2 root root 4096 Jan 30 00:00 root
drwxr-xr-x  1 root root 4096 Feb  6 07:49 run
drwxr-xr-x  2 root root 4096 Jan 30 00:00 sbin
drwxr-xr-x  2 root root 4096 Jan 30 00:00 srv
dr-xr-xr-x 13 root root    0 Feb  6 07:49 sys
drwxrwxrwt  1 root root 4096 Feb  2 08:06 tmp
drwxr-xr-x  1 root root 4096 Jan 30 00:00 usr
drwxr-xr-x  1 root root 4096 Jan 30 00:00 var
root@d4267c461c6c:/# 

这就是容器里的 结构

下面画个图理解

在这里插入图片描述

5.nginx 映射一个 别的页面

emmm 准备一个html： 给他映射到  我git博客上
-------------------------------
<!DOCTYPE html>
<html>
<head>
<title>Welcome to nginx!</title>
<style>
    body {
        width: 35em;
        margin: 0 auto;
        font-family: Tahoma, Verdana, Arial, sans-serif;
    }
</style>
</head>
<body>

<p id="p1">Hello Funs</p>
<a href="https://sxwanggit126.github.io/archives/">my blog</a>
</body>
</html>

准备：
[root@hadoop101 nginx]# vim doublehappy.html
<!DOCTYPE html>
<html>
  ead>
▽title>Welcome to nginx!</title>
<style>
    body {
        width: 35em;
        margin: 0 auto;
        font-family: Tahoma, Verdana, Arial, sans-serif;
    }
</style>
</head>
<body>

<p id="p1">Hello Funs</p>
<a href="https://sxwanggit126.github.io/archives/">www.ruozedata.com</a>
</body>
</html>
~                                                                                                                 
~                                                                                                                 
"doublehappy.html" [New] 18L, 330C written                                                      
[root@hadoop101 nginx]# vim doublehappy.html
<!DOCTYPE html>
<html>
<head>
<title>Welcome to nginx!</title>
<style>
    body {
▽       width: 35em;
        margin: 0 auto;
        font-family: Tahoma, Verdana, Arial, sans-serif;
    }
</style>
</head>
<body>

<p id="p1">Hello Funs</p>
<a href="https://sxwanggit126.github.io/archives/">www.ruozedata.com</a>
</body>
</html>
~                                                                                                                 
~                                                                                                                 
[root@hadoop101 nginx]# ll
total 4
-rw-r--r-- 1 root root 330 Feb  6 16:16 doublehappy.html
[root@hadoop101 nginx]# mv doublehappy.html  index.html
[root@hadoop101 nginx]# vim index.html 
<!DOCTYPE html>
<html>
<head>
<title>Welcome to nginx!</title>
<style>
    body {
        width: 35em;
        margin: 0 auto;
        font-family: Tahoma, Verdana, Arial, sans-serif;
    }
</style>
</head>
<body>

<p id="p1">Hello Funs</p>
<a href="https://sxwanggit126.github.io/archives/">www.ruozedata.com</a>
</body>
</html>
~                                                                                                                 
~                                                                                                                 
[root@hadoop101 nginx]# pwd
/root/nginx
[root@hadoop101 nginx]# 

默认 是 index.xml


命令：我们换一个 nginx  同时 换一个 端口号

docker run \
--name doublehappy-nginx2 \
-v /root/nginx:/usr/share/nginx/html:ro \
-d -p 8081:80 \
nginx:latest


-v：指的是 将本地的文件夹映射到 容器里面的文件夹
ro：指的是 你挂载的文件 是否可读和可写  r可读  w可写  即：rw


[root@hadoop101 nginx]# docker ps
CONTAINER ID        IMAGE               COMMAND                  CREATED             STATUS              PORTS                  NAMES
7fbe515cc1ee        nginx:latest        "nginx -g 'daemon of…"   7 seconds ago       Up 6 seconds        0.0.0.0:8081->80/tcp   doublehappy-nginx2
d4267c461c6c        nginx:latest        "nginx -g 'daemon of…"   30 minutes ago      Up 30 minutes       0.0.0.0:8080->80/tcp   doublehappy-nginx1
[root@hadoop101 nginx]#

在这里插入图片描述
点击链接即可进入映射的url

进入 容器：
[root@hadoop101 nginx]# docker exec -it doublehappy-nginx2 /bin/bash
root@7fbe515cc1ee:/# 

注意：
/bin/bash  指的是 以什么进行 解析 
因为 这个容器 相当于 运行一个 小的 linux 以 bash来进行解析


root@7fbe515cc1ee:/# ls -l /usr/share/nginx/html
total 4
-rw-r--r-- 1 root root 320 Feb  6 08:23 index.html
root@7fbe515cc1ee:/# cat /usr/share/nginx/html/index.html
<!DOCTYPE html>
<html>
<head>
<title>Welcome to nginx!</title>
<style>
    body {
        width: 35em;
        margin: 0 auto;
        font-family: Tahoma, Verdana, Arial, sans-serif;
    }
</style>
</head>
<body>

<p id="p1">Hello Funs</p>
<a href="https://sxwanggit126.github.io/archives/">my blog</a>
</body>
</html>
root@7fbe515cc1ee:/#

总结：
docker 目的就是 简化 

container 容器实例 其实就是一个ps的进程 containerd-shim，那么只能运行一个服务
这句话什么意思呢？
	就是 加入你要部署一个 伪分布式的 Hadoop
	nn，dn 这两个 都各自是一个进程 
	也就是说 你用docker 需要 部署两次 一个是 nn 一个是 dn

Mysql

[root@hadoop101 ~]# docker pull mysql:5.7
5.7: Pulling from library/mysql
619014d83c02: Pull complete 
9ced578c3a5f: Pull complete 
731f6e13d8ea: Pull complete 
3c183de42679: Pull complete 
6de69b5c2f3c: Pull complete 
00f0a4086406: Pull complete 
84d93aea836d: Pull complete 
e2dd1f3e5a3d: Pull complete 
98fb37478ee6: Pull complete 
57eb21662ebf: Pull complete 
e95057f0a050: Pull complete 
Digest: sha256:cf6899e980c38071a78ded028de40e65286bfbbb746b97617ac4c9a84c4e812d
Status: Downloaded newer image for mysql:5.7
docker.io/library/mysql:5.7
[root@hadoop101 ~]# 



[root@hadoop101 ~]# docker images
REPOSITORY          TAG                 IMAGE ID            CREATED             SIZE
nginx               latest              2073e0bcb60e        4 days ago          127MB
mysql               5.7                 c4f186b9e038        4 days ago          435MB
apache/kudu         latest              e1c6e5b62086        2 months ago        476MB
hello-world         latest              fce289e99eb9        13 months ago       1.84kB
kunickiaj/kudu      latest              c30c95f5fe58        17 months ago       896MB
docker/compose      1.8.0               89188432ef03        3 years ago         59.1MB
[root@hadoop101 ~]#

在这里插入图片描述

3.如何去运行  看上面图片

docker run \
--name doublehappy-mysql1 \
-e MYSQL_ROOT_PASSWORD=123456 \
-d mysql:5.7

注意：
-e 指的是 环境变量
 这个密码怎么传过去的呢？一会解释


[root@hadoop101 ~]# docker run \
> --name doublehappy-mysql1 \
> -e MYSQL_ROOT_PASSWORD=123456 \
> -d mysql:5.7
ed9c2cb2591cc8fa85af1c14497bed28c4da9b8c94e94e50e5c5b89ff93154cc
[root@hadoop101 ~]# 


[root@hadoop101 ~]# docker ps
CONTAINER ID        IMAGE               COMMAND                  CREATED             STATUS              PORTS                  NAMES
ed9c2cb2591c        mysql:5.7           "docker-entrypoint.s…"   45 seconds ago      Up 44 seconds       3306/tcp, 33060/tcp    doublehappy-mysql1
7fbe515cc1ee        nginx:latest        "nginx -g 'daemon of…"   29 minutes ago      Up 29 minutes       0.0.0.0:8081->80/tcp   doublehappy-nginx2
d4267c461c6c        nginx:latest        "nginx -g 'daemon of…"   About an hour ago   Up About an hour    0.0.0.0:8080->80/tcp   doublehappy-nginx1
[root@hadoop101 ~]# 


3306/tcp, 33060/tcp:
	表示多个端口 对外提供 映射服务的

这个没有箭头 说明外面是 访问不了的 


[root@hadoop101 ~]# docker exec -it doublehappy-mysql1 /bin/bash
root@ed9c2cb2591c:/# which ps
root@ed9c2cb2591c:/# 
root@ed9c2cb2591c:/# 
root@ed9c2cb2591c:/# mysql -uroot -p123456
mysql: [Warning] Using a password on the command line interface can be insecure.
Welcome to the MySQL monitor.  Commands end with ; or \g.
Your MySQL connection id is 2
Server version: 5.7.29 MySQL Community Server (GPL)

Copyright (c) 2000, 2020, Oracle and/or its affiliates. All rights reserved.

Oracle is a registered trademark of Oracle Corporation and/or its
affiliates. Other names may be trademarks of their respective
owners.

Type 'help;' or '\h' for help. Type '\c' to clear the current input statement.

mysql>
 +--------------------+
| Database           |
+--------------------+
| information_schema |
| mysql              |
| performance_schema |
| sys                |
+--------------------+
4 rows in set (0.00 sec)

mysql> 


如何对外访问？？
  之前的 nginx 是 命令 加一个 -p

docker run \
--name doublehappy-mysql2 \
-e MYSQL_ROOT_PASSWORD=123456 \
-p 33060:3306 \
-d mysql:5.7


[root@hadoop101 ~]# docker run \
> --name doublehappy-mysql2 \
> -e MYSQL_ROOT_PASSWORD=123456 \
> -p 33060:3306 \
> -d mysql:5.7
b66ce92bfa4f5c53a588a41e015bb7b045eaa602df5cf0d98c40bec7c3bd6571
[root@hadoop101 ~]# 


[root@hadoop101 ~]# docker ps
CONTAINER ID        IMAGE               COMMAND                  CREATED             STATUS              PORTS                                NAMES
b66ce92bfa4f        mysql:5.7           "docker-entrypoint.s…"   13 seconds ago      Up 12 seconds       33060/tcp, 0.0.0.0:33060->3306/tcp   doublehappy-mysql2
ed9c2cb2591c        mysql:5.7           "docker-entrypoint.s…"   6 minutes ago       Up 6 minutes        3306/tcp, 33060/tcp                  doublehappy-mysql1
7fbe515cc1ee        nginx:latest        "nginx -g 'daemon of…"   35 minutes ago      Up 35 minutes       0.0.0.0:8081->80/tcp                 doublehappy-nginx2
d4267c461c6c        nginx:latest        "nginx -g 'daemon of…"   About an hour ago   Up About an hour    0.0.0.0:8080->80/tcp                 doublehappy-nginx1
[root@hadoop101 ~]# 

说明：
33060/tcp, 0.0.0.0:33060->3306/tcp  映射成功 

ok 我们 用 海迪链接一下

在这里插入图片描述

[root@hadoop101 ~]# ps -ef | grep docker
root      6599     1  0 14:13 ?        00:00:19 /usr/bin/dockerd -H fd:// --containerd=/run/containerd/containerd.sock
root     11669  6599  0 15:49 ?        00:00:00 /usr/bin/docker-proxy -proto tcp -host-ip 0.0.0.0 -host-port 8080 -container-ip 172.17.0.2 -container-port 80
root     11701  6598  0 15:49 ?        00:00:00 containerd-shim -namespace moby -workdir /var/lib/containerd/io.containerd.runtime.v1.linux/moby/d4267c461c6cb0cfaeac3a2e4037f6ad1fbb57bfe31264888ff1e4ca220e5f8b -address /run/containerd/containerd.sock -containerd-binary /usr/bin/containerd -runtime-root /var/run/docker/runtime-runc
root     13582  3143  0 16:01 pts/1    00:00:00 docker exec -it d4267c461c6c /bin/bash
root     13909  6599  0 16:20 ?        00:00:00 /usr/bin/docker-proxy -proto tcp -host-ip 0.0.0.0 -host-port 8081 -container-ip 172.17.0.3 -container-port 80
root     13925  6598  0 16:20 ?        00:00:00 containerd-shim -namespace moby -workdir /var/lib/containerd/io.containerd.runtime.v1.linux/moby/7fbe515cc1ee97390b66f686f0119232c9f329898b9f57bebabdfd89f23fd641 -address /run/containerd/containerd.sock -containerd-binary /usr/bin/containerd -runtime-root /var/run/docker/runtime-runc
root     14128 13798  0 16:25 pts/0    00:00:00 docker exec -it doublehappy-nginx2 /bin/bash
root     16274  6598  0 16:49 ?        00:00:00 containerd-shim -namespace moby -workdir /var/lib/containerd/io.containerd.runtime.v1.linux/moby/ed9c2cb2591cc8fa85af1c14497bed28c4da9b8c94e94e50e5c5b89ff93154cc -address /run/containerd/containerd.sock -containerd-binary /usr/bin/containerd -runtime-root /var/run/docker/runtime-runc
root     16786  6599  0 16:55 ?        00:00:00 /usr/bin/docker-proxy -proto tcp -host-ip 0.0.0.0 -host-port 33060 -container-ip 172.17.0.5 -container-port 3306
root     16810  6598  0 16:55 ?        00:00:00 containerd-shim -namespace moby -workdir /var/lib/containerd/io.containerd.runtime.v1.linux/moby/b66ce92bfa4f5c53a588a41e015bb7b045eaa602df5cf0d98c40bec7c3bd6571 -address /run/containerd/containerd.sock -containerd-binary /usr/bin/containerd -runtime-root /var/run/docker/runtime-runc
root     17188 14323  0 17:04 pts/2    00:00:00 grep --color=auto docker
[root@hadoop101 ~]# 



runtime-runc  有四个 


我们进程 运行了 四个 正确

注意：
	是个进程 
	那么 mysql 的数据存放在哪里呢？
	会不会随着进程的kill 而数据丢失呢？

docker 架构

在这里插入图片描述
架构

按照图片 线 走

docker 如何构建自己的镜像

以官方 mysql为例
 mysql Dockerfile 脚本：相当于入口

1.找

注意：
	要构建我们的 Dockerfile    就是把官网的 拿来

[root@hadoop101 ~]# mkdir mysql5.7
[root@hadoop101 ~]# cd mysql5.7/
[root@hadoop101 mysql5.7]# vi Dockerfile
FROM debian:stretch-slim

RUN groupadd -r mysql && useradd -r -g mysql mysql

RUN apt-get update && apt-get install -y --no-install-recommends gnupg dirmngr && rm -rf /var/lib/apt/lists/*

# add gosu for easy step-down from root
ENV GOSU_VERSION 1.7
RUN set -x \
        && export GNUPGHOME="$(mktemp -d)" \
        && gpg --batch --verify /usr/local/bin/gosu.asc /usr/local/bin/gosu \
        && gpgconf --kill all \
        && rm -rf "$GNUPGHOME" /usr/local/bin/gosu.asc \
        && chmod +x /usr/local/bin/gosu \
        && gosu nobody true \
        && apt-get purge -y --auto-remove ca-certificates wget

RUN mkdir /docker-entrypoint-initdb.d

RUN apt-get update && apt-get install -y --no-install-recommends \
# for MYSQL_RANDOM_ROOT_PASSWORD
                pwgen \
# for mysql_ssl_rsa_setup
                openssl \
# File::Basename
# File::Copy
# Sys::Hostname
        && rm -rf /var/lib/apt/lists/*

RUN set -ex; \
# gpg: key 5072E1F5: public key "MySQL Release Engineering <mysql-build@oss.oracle.com>" imported
        key='A4A9406876FCBD3C456770C88C718D3B5072E1F5'; \
        export GNUPGHOME="$(mktemp -d)"; \
        gpg --batch --keyserver ha.pool.sks-keyservers.net --recv-keys "$key"; \
        gpg --batch --export "$key" > /etc/apt/trusted.gpg.d/mysql.gpg; \
        gpgconf --kill all; \
        rm -rf "$GNUPGHOME"; \
ENV MYSQL_MAJOR 5.7
ENV MYSQL_VERSION 5.7.29-1debian9

RUN echo "deb http://repo.mysql.com/apt/debian/ stretch mysql-${MYSQL_MAJOR}" > /etc/apt/sources.list.d/mysql.list

# also, we set debconf keys to make APT a little quieter
RUN { \
                echo mysql-community-server mysql-community-server/data-dir select ''; \
                echo mysql-community-server mysql-community-server/root-pass password ''; \
                echo mysql-community-server mysql-community-server/re-root-pass password ''; \
                echo mysql-community-server mysql-community-server/remove-test-db select false; \
        } | debconf-set-selections \
        && apt-get update && apt-get install -y mysql-server="${MYSQL_VERSION}" && rm -rf /var/lib/apt/lists/* \
        && rm -rf /var/lib/mysql && mkdir -p /var/lib/mysql /var/run/mysqld \
        && chown -R mysql:mysql /var/lib/mysql /var/run/mysqld \
# ensure that /var/run/mysqld (used for socket and lock files) is writable regardless of the UID our mysqld instan
ce ends up having at runtime
        && chmod 777 /var/run/mysqld \
# comment out a few problematic configuration values
        && find /etc/mysql/ -name '*.cnf' -print0 \
                | xargs -0 grep -lZE '^(bind-address|log)' \
                | xargs -rt -0 sed -Ei 's/^(bind-address|log)/#&/' \
# don't reverse lookup hostnames, they are usually another container
        && echo '[mysqld]\nskip-host-cache\nskip-name-resolve' > /etc/mysql/conf.d/docker.cnf

VOLUME /var/lib/mysql

COPY docker-entrypoint.sh /usr/local/bin/
RUN ln -s usr/local/bin/docker-entrypoint.sh /entrypoint.sh # backwards compat
ENTRYPOINT ["docker-entrypoint.sh"]

EXPOSE 3306 33060
CMD ["mysqld"]
"Dockerfile" [New] 80L, 3803C written
[root@hadoop101 mysql5.7]# 	


注意：
	一般Dockerfile 都有 shell脚本 看看 Dockerfile    docker-entrypoint.sh
 这个脚本就是入口

把这个脚本也做好 

[root@hadoop101 mysql5.7]# ll
total 20
-rw-r--r-- 1 root root 12713 Feb  6 17:23 docker-entrypoint.sh
-rw-r--r-- 1 root root  3803 Feb  6 17:18 Dockerfile
[root@hadoop101 mysql5.7]# 


2.命令构建

[root@hadoop101 mysql5.7]# ll
total 20
-rw-r--r-- 1 root root 12713 Feb  6 17:23 docker-entrypoint.sh
-rw-r--r-- 1 root root  3803 Feb  6 17:18 Dockerfile

[root@hadoop101 mysql5.7]# docker build ./ -t mysql:doublehappy5.7
Sending build context to Docker daemon  18.94kB
Step 1/18 : FROM debian:stretch-slim



注意：
	建议 用国外的 服务器 可以翻墙的 要不然真的很慢

3.解读Dockerfile：
FROM 基础镜像
	eg:
	FROM debian:stretch-slim
		系统名字: 镜像的名字
		注意：debian 这个系统比较轻量级 centos 比较重 所以  debian 最终构建的 镜像比较小


RUN 在基础镜像的Linux操作命令
	eg：RUN groupadd -r mysql && useradd -r -g mysql mysql

apt-get ==》yum 

ENV 环境变量
    k=v  $k

VOLUME 定义匿名卷      !!!!  就是 文件的目录 这个很重要

COPY  复制文本
      源路径  目标路径


ENTRYPOINT ["docker-entrypoint.sh"]
           设定容器启动时第一个运行的命令及参数
			也就是说 这个shell 脚本里面 可以接收外部的 参数 

EXPOSE 3306 33060  
      仅作为容器的服务使用什么端口号，且并不会主动在宿主Linux上自动映射端口

CMD ["mysqld"]
           运行服务

总结：
	毕竟 这是官网的
	你生产上 你要根据你自己的需求 来构建镜像 
	Dockfile 如何编写 和 shell 脚本 都是要你自己来写哒

shell 脚本：

/bin/bash

f2() {
 k2=v2
}

f1() {
 echo $k2
}

_main() {

}

注意：
	shell脚本是 从上往下加载的

停止

[root@hadoop101 ~]# docker ps
CONTAINER ID        IMAGE               COMMAND                  CREATED             STATUS              PORTS                                NAMES
3b4168987e1b        657af8c03255        "/bin/sh -c 'set -x …"   5 minutes ago       Up 5 minutes                                             agitated_sammet
b66ce92bfa4f        mysql:5.7           "docker-entrypoint.s…"   2 hours ago         Up 2 hours          33060/tcp, 0.0.0.0:33060->3306/tcp   doublehappy-mysql2
ed9c2cb2591c        mysql:5.7           "docker-entrypoint.s…"   2 hours ago         Up 2 hours          3306/tcp, 33060/tcp                  doublehappy-mysql1
7fbe515cc1ee        nginx:latest        "nginx -g 'daemon of…"   3 hours ago         Up 3 hours          0.0.0.0:8081->80/tcp                 doublehappy-nginx2
d4267c461c6c        nginx:latest        "nginx -g 'daemon of…"   3 hours ago         Up 3 hours          0.0.0.0:8080->80/tcp                 doublehappy-nginx1


[root@hadoop101 ~]# docker stop doublehappy-mysql1
doublehappy-mysql1
[root@hadoop101 ~]# docker ps
CONTAINER ID        IMAGE               COMMAND                  CREATED             STATUS              PORTS                                NAMES
3b4168987e1b        657af8c03255        "/bin/sh -c 'set -x …"   6 minutes ago       Up 6 minutes                                             agitated_sammet
b66ce92bfa4f        mysql:5.7           "docker-entrypoint.s…"   2 hours ago         Up 2 hours          33060/tcp, 0.0.0.0:33060->3306/tcp   doublehappy-mysql2
7fbe515cc1ee        nginx:latest        "nginx -g 'daemon of…"   3 hours ago         Up 3 hours          0.0.0.0:8081->80/tcp                 doublehappy-nginx2
d4267c461c6c        nginx:latest        "nginx -g 'daemon of…"   3 hours ago         Up 3 hours          0.0.0.0:8080->80/tcp                 doublehappy-nginx1
[root@hadoop101 ~]#

[root@hadoop101 ~]# docker --help

Usage:  docker [OPTIONS] COMMAND

A self-sufficient runtime for containers

Options:
      --config string      Location of client config files (default "/root/.docker")
  -c, --context string     Name of the context to use to connect to the daemon (overrides DOCKER_HOST env
                           var and default context set with "docker context use")
  -D, --debug              Enable debug mode
  -H, --host list          Daemon socket(s) to connect to
  -l, --log-level string   Set the logging level ("debug"|"info"|"warn"|"error"|"fatal") (default "info")
      --tls                Use TLS; implied by --tlsverify
      --tlscacert string   Trust certs signed only by this CA (default "/root/.docker/ca.pem")
      --tlscert string     Path to TLS certificate file (default "/root/.docker/cert.pem")
      --tlskey string      Path to TLS key file (default "/root/.docker/key.pem")
      --tlsverify          Use TLS and verify the remote
  -v, --version            Print version information and quit

Management Commands:
  builder     Manage builds
  config      Manage Docker configs
  container   Manage containers
  context     Manage contexts
  engine      Manage the docker engine
  image       Manage images
  network     Manage networks
  node        Manage Swarm nodes
  plugin      Manage plugins
  secret      Manage Docker secrets
  service     Manage services
  stack       Manage Docker stacks
  swarm       Manage Swarm
  system      Manage Docker
  trust       Manage trust on Docker images
  volume      Manage volumes

Commands:
  attach      Attach local standard input, output, and error streams to a running container
  build       Build an image from a Dockerfile
  commit      Create a new image from a container's changes
  cp          Copy files/folders between a container and the local filesystem
  create      Create a new container
  diff        Inspect changes to files or directories on a container's filesystem
  events      Get real time events from the server
  exec        Run a command in a running container
  export      Export a container's filesystem as a tar archive
  history     Show the history of an image
  images      List images
  import      Import the contents from a tarball to create a filesystem image
  info        Display system-wide information
  inspect     Return low-level information on Docker objects
  kill        Kill one or more running containers
  load        Load an image from a tar archive or STDIN
  login       Log in to a Docker registry
  logout      Log out from a Docker registry
  logs        Fetch the logs of a container
  pause       Pause all processes within one or more containers
  port        List port mappings or a specific mapping for the container
  ps          List containers
  pull        Pull an image or a repository from a registry
  push        Push an image or a repository to a registry
  rename      Rename a container
  restart     Restart one or more containers
  rm          Remove one or more containers
  rmi         Remove one or more images
  run         Run a command in a new container
  save        Save one or more images to a tar archive (streamed to STDOUT by default)
  search      Search the Docker Hub for images
  start       Start one or more stopped containers
  stats       Display a live stream of container(s) resource usage statistics
  stop        Stop one or more running containers
  tag         Create a tag TARGET_IMAGE that refers to SOURCE_IMAGE
  top         Display the running processes of a container
  unpause     Unpause all processes within one or more containers
  update      Update configuration of one or more containers
  version     Show the Docker version information
  wait        Block until one or more containers stop, then print their exit codes

Run 'docker COMMAND --help' for more information on a command.
[root@hadoop101 ~]#

k8s-Spark-doublehappy

2020-04-27

Spark ON K8S

Spark on K8s
在这里插入图片描述

k8s 数据存储是 etcd（生产上 这个 etcd 要做集群的）

The Kubernetes scheduler is currently experimental. 
spark on k8s 是实验阶段 所以生产上 没有使用 只是测试 

环境准备：
1.下载Spark
wget https://archive.apache.org/dist/spark/spark-2.4.3/spark-2.4.3-bin-hadoop2.6.tgz
tar -xzvf spark-2.4.3-bin-hadoop2.6.tgz -C /opt/
mv /opt/spark-2.4.3-bin-hadoop2.6 /opt/spark   
cd  /opt/spark

注意：一定是 /opt/spark 目录下 ！！！
[root@container01 spark]# pwd
/opt/spark
[root@container01 spark]# ll
total 128
drwxr-xr-x 2 1000 1000  4096 May  1  2019 bin
drwxr-xr-x 2 1000 1000  4096 May  1  2019 conf
drwxr-xr-x 5 1000 1000  4096 May  1  2019 data
drwxr-xr-x 4 1000 1000  4096 May  1  2019 examples
drwxr-xr-x 2 1000 1000 12288 May  1  2019 jars
drwxr-xr-x 4 1000 1000  4096 May  1  2019 kubernetes
-rw-r--r-- 1 1000 1000 21316 May  1  2019 LICENSE
drwxr-xr-x 2 1000 1000  4096 May  1  2019 licenses
-rw-r--r-- 1 1000 1000 42919 May  1  2019 NOTICE
drwxr-xr-x 7 1000 1000  4096 May  1  2019 python
drwxr-xr-x 3 1000 1000  4096 May  1  2019 R
-rw-r--r-- 1 1000 1000  3952 May  1  2019 README.md
-rw-r--r-- 1 1000 1000   164 May  1  2019 RELEASE
drwxr-xr-x 2 1000 1000  4096 May  1  2019 sbin
drwxr-xr-x 2 1000 1000  4096 May  1  2019 yarn
[root@container01 spark]#


2.构建spark镜像
./bin/docker-image-tool.sh -r 172.21.230.89/doublehappy -t v2.4.3 build
./bin/docker-image-tool.sh -r 172.21.230.89/doublehappy -t v2.4.3 push

[root@container01 spark]# ./bin/docker-image-tool.sh -r 172.21.230.89/doublehappy -t v2.4.3 build
Sending build context to Docker daemon  259.1MB
Step 1/15 : FROM openjdk:8-alpine
8-alpine: Pulling from library/openjdk
e7c96db7181b: Pull complete 
f910a506b6cb: Pull complete 
c2274a1a0e27: Pull complete 
Digest: sha256:94792824df2df33402f201713f932b58cb9de94a0cd524164a0f2283343547b3
Status: Downloaded newer image for openjdk:8-alpine
 ---> a3562aa0b991
Step 2/15 : ARG spark_jars=jars
 ---> Running in 1ed1dcbc748b
Removing intermediate container 1ed1dcbc748b
 ---> 4bf840b35abc
Step 3/15 : ARG img_path=kubernetes/dockerfiles
 ---> Running in 2e0a10a5eeeb
Removing intermediate container 2e0a10a5eeeb
 ---> e7b176887b3d
Step 4/15 : ARG k8s_tests=kubernetes/tests
 ---> Running in 5d78a64b22a9
Removing intermediate container 5d78a64b22a9
 ---> 7dda4f6f30d8
Step 5/15 : RUN set -ex &&     apk upgrade --no-cache &&     apk add --no-cache bash tini libc6-compat linux-pam nss &&     mkdir -p /opt/spark &&     mkdir -p /opt/spark/work-dir &&     touch /opt/spark/RELEASE &&     rm /bin/sh &&     ln -sv /bin/bash /bin/sh &&     echo "auth required pam_wheel.so use_uid" >> /etc/pam.d/su &&     chgrp root /etc/passwd && chmod ug+rw /etc/passwd
 ---> Running in 52cff938c375
+ apk upgrade --no-cache
fetch http://dl-cdn.alpinelinux.org/alpine/v3.9/main/x86_64/APKINDEX.tar.gz
fetch http://dl-cdn.alpinelinux.org/alpine/v3.9/community/x86_64/APKINDEX.tar.gz
(1/11) Upgrading musl (1.1.20-r4 -> 1.1.20-r5)
(2/11) Upgrading libcrypto1.1 (1.1.1b-r1 -> 1.1.1d-r2)
(3/11) Upgrading libssl1.1 (1.1.1b-r1 -> 1.1.1d-r2)
(4/11) Upgrading musl-utils (1.1.20-r4 -> 1.1.20-r5)
(5/11) Upgrading libtasn1 (4.13-r0 -> 4.14-r0)
(6/11) Upgrading sqlite-libs (3.26.0-r3 -> 3.28.0-r2)
(7/11) Upgrading libbz2 (1.0.6-r6 -> 1.0.6-r7)
(8/11) Upgrading libjpeg-turbo (1.5.3-r4 -> 1.5.3-r6)
(9/11) Upgrading libcom_err (1.44.5-r0 -> 1.44.5-r2)
(10/11) Upgrading openjdk8-jre-base (8.212.04-r0 -> 8.242.08-r0)
(11/11) Upgrading openjdk8-jre (8.212.04-r0 -> 8.242.08-r0)
Executing busybox-1.29.3-r10.trigger
Executing ca-certificates-20190108-r0.trigger
Executing java-common-0.1-r0.trigger
OK: 103 MiB in 54 packages
+ apk add --no-cache bash tini libc6-compat linux-pam nss
fetch http://dl-cdn.alpinelinux.org/alpine/v3.9/main/x86_64/APKINDEX.tar.gz
fetch http://dl-cdn.alpinelinux.org/alpine/v3.9/community/x86_64/APKINDEX.tar.gz
(1/8) Installing ncurses-terminfo-base (6.1_p20190105-r0)
(2/8) Installing ncurses-terminfo (6.1_p20190105-r0)
(3/8) Installing ncurses-libs (6.1_p20190105-r0)
(4/8) Installing readline (7.0.003-r1)
(5/8) Installing bash (4.4.19-r1)
Executing bash-4.4.19-r1.post-install
(6/8) Installing libc6-compat (1.1.20-r5)
(7/8) Installing linux-pam (1.3.0-r0)
(8/8) Installing tini (0.18.0-r0)
Executing busybox-1.29.3-r10.trigger
OK: 113 MiB in 62 packages
+ mkdir -p /opt/spark
+ mkdir -p /opt/spark/work-dir
+ touch /opt/spark/RELEASE
+ rm /bin/sh
+ ln -sv /bin/bash /bin/sh
'/bin/sh' -> '/bin/bash'
+ echo 'auth required pam_wheel.so use_uid'
+ chgrp root /etc/passwd
+ chmod ug+rw /etc/passwd
Removing intermediate container 52cff938c375
 ---> afcf7dcb4c3f
Step 6/15 : COPY ${spark_jars} /opt/spark/jars
 ---> 89385fbda28a
Step 7/15 : COPY bin /opt/spark/bin
 ---> aa3513116505
Step 8/15 : COPY sbin /opt/spark/sbin
 ---> 47bfb4c55137
Step 9/15 : COPY ${img_path}/spark/entrypoint.sh /opt/
 ---> 4e1505375576
Step 10/15 : COPY examples /opt/spark/examples
 ---> 516d18d76b12
Step 11/15 : COPY ${k8s_tests} /opt/spark/tests
 ---> 9f89a7c63732
Step 12/15 : COPY data /opt/spark/data
 ---> 4f630068f4e7
Step 13/15 : ENV SPARK_HOME /opt/spark
 ---> Running in 5a3c23422bf6
Removing intermediate container 5a3c23422bf6
 ---> f2b96e5d55df
Step 14/15 : WORKDIR /opt/spark/work-dir
 ---> Running in 7579f5960498
Removing intermediate container 7579f5960498
 ---> c2b2f4e804be
Step 15/15 : ENTRYPOINT [ "/opt/entrypoint.sh" ]
 ---> Running in 54c8dd48bf2c
Removing intermediate container 54c8dd48bf2c
 ---> 7d895d565b9d
Successfully built 7d895d565b9d
Successfully tagged 172.21.230.89/doublehappy/spark:v2.4.3
Sending build context to Docker daemon  259.1MB
Step 1/9 : ARG base_img
Step 2/9 : FROM $base_img
 ---> 7d895d565b9d
Step 3/9 : WORKDIR /
 ---> Running in 676ac68db202
Removing intermediate container 676ac68db202
 ---> 72e742d2ebe3
Step 4/9 : RUN mkdir ${SPARK_HOME}/python
 ---> Running in e7792928115a
Removing intermediate container e7792928115a
 ---> 00148d7dc2ff
Step 5/9 : RUN apk add --no-cache python &&     apk add --no-cache python3 &&     python -m ensurepip &&     python3 -m ensurepip &&     rm -r /usr/lib/python*/ensurepip &&     pip install --upgrade pip setuptools &&     rm -r /root/.cache
 ---> Running in 959225826ce7
fetch http://dl-cdn.alpinelinux.org/alpine/v3.9/main/x86_64/APKINDEX.tar.gz
fetch http://dl-cdn.alpinelinux.org/alpine/v3.9/community/x86_64/APKINDEX.tar.gz
(1/3) Installing expat (2.2.8-r0)
(2/3) Installing gdbm (1.13-r1)
(3/3) Installing python2 (2.7.16-r2)
Executing busybox-1.29.3-r10.trigger
OK: 152 MiB in 65 packages
fetch http://dl-cdn.alpinelinux.org/alpine/v3.9/main/x86_64/APKINDEX.tar.gz
fetch http://dl-cdn.alpinelinux.org/alpine/v3.9/community/x86_64/APKINDEX.tar.gz
(1/2) Installing xz-libs (5.2.4-r0)
(2/2) Installing python3 (3.6.9-r2)
Executing busybox-1.29.3-r10.trigger
OK: 205 MiB in 67 packages
Looking in links: /tmp/tmpxKrRmR
Collecting setuptools
Collecting pip
Installing collected packages: setuptools, pip
Successfully installed pip-18.1 setuptools-40.6.2
Looking in links: /tmp/tmp83v437xu
Requirement already satisfied: setuptools in /usr/lib/python3.6/site-packages (40.6.2)
Requirement already satisfied: pip in /usr/lib/python3.6/site-packages (18.1)
Collecting pip
  Downloading https://files.pythonhosted.org/packages/54/0c/d01aa759fdc501a58f431eb594a17495f15b88da142ce14b5845662c13f3/pip-20.0.2-py2.py3-none-any.whl (1.4MB)
Collecting setuptools
  Downloading https://files.pythonhosted.org/packages/f9/d3/955738b20d3832dfa3cd3d9b07e29a8162edb480bf988332f5e6e48ca444/setuptools-44.0.0-py2.py3-none-any.whl (583kB)
Installing collected packages: pip, setuptools
  Found existing installation: pip 18.1
    Uninstalling pip-18.1:
      Successfully uninstalled pip-18.1
  Found existing installation: setuptools 40.6.2
    Uninstalling setuptools-40.6.2:
      Successfully uninstalled setuptools-40.6.2
Successfully installed pip-20.0.2 setuptools-44.0.0
Removing intermediate container 959225826ce7
 ---> d803cfb93fb7
Step 6/9 : COPY python/lib ${SPARK_HOME}/python/lib
 ---> a6ff2890f901
Step 7/9 : ENV PYTHONPATH ${SPARK_HOME}/python/lib/pyspark.zip:${SPARK_HOME}/python/lib/py4j-*.zip
 ---> Running in 239d8bd82251
Removing intermediate container 239d8bd82251
 ---> a3899eac9c76
Step 8/9 : WORKDIR /opt/spark/work-dir
 ---> Running in afa1f7f2fba0
Removing intermediate container afa1f7f2fba0
 ---> 0924048e54c9
Step 9/9 : ENTRYPOINT [ "/opt/entrypoint.sh" ]
 ---> Running in 3775107781f5
Removing intermediate container 3775107781f5
 ---> 0807413bcdcd
Successfully built 0807413bcdcd
Successfully tagged 172.21.230.89/doublehappy/spark-py:v2.4.3
Sending build context to Docker daemon  259.1MB
Step 1/9 : ARG base_img
Step 2/9 : FROM $base_img
 ---> 7d895d565b9d
Step 3/9 : WORKDIR /
 ---> Using cache
 ---> 72e742d2ebe3
Step 4/9 : RUN mkdir ${SPARK_HOME}/R
 ---> Running in 7e96151ab7ee
Removing intermediate container 7e96151ab7ee
 ---> 8a1fde912f50
Step 5/9 : RUN apk add --no-cache R R-dev
 ---> Running in f2a9cfb0ee09
fetch http://dl-cdn.alpinelinux.org/alpine/v3.9/main/x86_64/APKINDEX.tar.gz
fetch http://dl-cdn.alpinelinux.org/alpine/v3.9/community/x86_64/APKINDEX.tar.gz
(1/60) Installing R-mathlib (3.5.1-r1)
(2/60) Installing libice (1.0.9-r3)
(3/60) Installing libuuid (2.33-r0)
(4/60) Installing libsm (1.2.2-r2)
(5/60) Installing libxt (1.1.5-r2)
(6/60) Installing libxmu (1.1.2-r1)
(7/60) Installing expat (2.2.8-r0)
(8/60) Installing fontconfig (2.13.1-r0)
(9/60) Installing pixman (0.34.0-r6)
(10/60) Installing cairo (1.16.0-r1)
(11/60) Installing nghttp2-libs (1.35.1-r1)
(12/60) Installing libssh2 (1.9.0-r1)
(13/60) Installing libcurl (7.64.0-r3)
(14/60) Installing libintl (0.19.8.1-r4)
(15/60) Installing libblkid (2.33-r0)
(16/60) Installing libmount (2.33-r0)
(17/60) Installing pcre (8.42-r1)
(18/60) Installing glib (2.58.1-r3)
(19/60) Installing libgomp (8.3.0-r0)
(20/60) Installing icu-libs (62.1-r0)
(21/60) Installing xz-libs (5.2.4-r0)
(22/60) Installing libquadmath (8.3.0-r0)
(23/60) Installing libgfortran (8.3.0-r0)
(24/60) Installing openblas (0.3.3-r2)
(25/60) Installing libxft (2.3.2-r3)
(26/60) Installing fribidi (1.0.5-r1)
(27/60) Installing graphite2 (1.3.12-r1)
(28/60) Installing harfbuzz (2.2.0-r0)
(29/60) Installing pango (1.42.4-r1)
(30/60) Installing tiff (4.0.10-r3)
(31/60) Installing R (3.5.1-r1)
Executing R-3.5.1-r1.post-install
*
* If you want to install R packages from CRAN that contains native extensions,
* then you must also install R-dev.
*
(32/60) Installing binutils (2.31.1-r2)
(33/60) Installing gmp (6.1.2-r1)
(34/60) Installing isl (0.18-r0)
(35/60) Installing libatomic (8.3.0-r0)
(36/60) Installing mpfr3 (3.1.5-r1)
(37/60) Installing mpc1 (1.0.3-r1)
(38/60) Installing gcc (8.3.0-r0)
(39/60) Installing gfortran (8.3.0-r0)
(40/60) Installing icu (62.1-r0)
(41/60) Installing pkgconf (1.6.0-r0)
(42/60) Installing icu-dev (62.1-r0)
(43/60) Installing zlib-dev (1.2.11-r1)
(44/60) Installing libpng-dev (1.6.37-r0)
(45/60) Installing make (4.2.1-r2)
(46/60) Installing openblas-ilp64 (0.3.3-r2)
(47/60) Installing openblas-dev (0.3.3-r2)
(48/60) Installing libpcre16 (8.42-r1)
(49/60) Installing libpcre32 (8.42-r1)
(50/60) Installing libpcrecpp (8.42-r1)
(51/60) Installing pcre-dev (8.42-r1)
(52/60) Installing libhistory (7.0.003-r1)
(53/60) Installing readline-dev (7.0.003-r1)
(54/60) Installing xz-dev (5.2.4-r0)
(55/60) Installing bzip2-dev (1.0.6-r7)
(56/60) Installing openssl-dev (1.1.1d-r2)
(57/60) Installing libssh2-dev (1.9.0-r1)
(58/60) Installing nghttp2-dev (1.35.1-r1)
(59/60) Installing curl-dev (7.64.0-r3)
(60/60) Installing R-dev (3.5.1-r1)
Executing busybox-1.29.3-r10.trigger
Executing glib-2.58.1-r3.trigger
OK: 483 MiB in 122 packages
Removing intermediate container f2a9cfb0ee09
 ---> 6ba3e96945f1
Step 6/9 : COPY R ${SPARK_HOME}/R
 ---> 53c866b24bc7
Step 7/9 : ENV R_HOME /usr/lib/R
 ---> Running in 353fcf7a3195
Removing intermediate container 353fcf7a3195
 ---> 15f73ab74099
Step 8/9 : WORKDIR /opt/spark/work-dir
 ---> Running in 6f8f14f50dd1
Removing intermediate container 6f8f14f50dd1
 ---> f9ba7e7be95f
Step 9/9 : ENTRYPOINT [ "/opt/entrypoint.sh" ]
 ---> Running in b45e8b5b55ca
Removing intermediate container b45e8b5b55ca
 ---> e77ac91a3898
Successfully built e77ac91a3898
Successfully tagged 172.21.230.89/doublehappy/spark-r:v2.4.3
[root@container01 spark]# 


查看一下：

[root@container01 ~]# docker images
REPOSITORY                                                        TAG                              IMAGE ID            CREATED             SIZE
<none>                                                            <none>                           8a1fde912f50        12 seconds ago      371MB
172.21.230.89/doublehappy/spark-py                                v2.4.3                           0807413bcdcd        15 seconds ago      462MB
172.21.230.89/doublehappy/spark                                   v2.4.3                           7d895d565b9d        36 seconds ago      371MB
mysql                                                             doublehappy5.7                   5c68aaadf900        4 days ago          434MB
debian                                                            stretch-slim                     5174edbf8c39        9 days ago          55.3MB
goharbor/chartmuseum-photon                                       v0.9.0-v1.10.0                   c8c98d2ac22b        2 months ago        128MB
goharbor/harbor-migrator                                          v1.10.0                          9dd0c9228c79        2 months ago        362MB
goharbor/redis-photon                                             v1.10.0                          6df66e5c1ca7        2 months ago        111MB
goharbor/clair-adapter-photon                                     v1.0.1-v1.10.0                   4bdc58f374c2        2 months ago        61.6MB
goharbor/clair-photon                                             v2.1.1-v1.10.0                   435d235c5000        2 months ago        171MB
goharbor/notary-server-photon                                     v0.6.1-v1.10.0                   c7934c32e873        2 months ago        143MB
goharbor/notary-signer-photon                                     v0.6.1-v1.10.0                   bb1a762155ab        2 months ago        140MB
goharbor/harbor-registryctl                                       v1.10.0                          c550280445e6        2 months ago        104MB
goharbor/registry-photon                                          v2.7.1-patch-2819-2553-v1.10.0   2115e08fa399        2 months ago        86.5MB
goharbor/nginx-photon                                             v1.10.0                          f7ed614c3abc        2 months ago        44MB
goharbor/harbor-log                                               v1.10.0                          fb15f6772e9a        2 months ago        82.3MB
goharbor/harbor-jobservice                                        v1.10.0                          d6d4f2b125f6        2 months ago        142MB
goharbor/harbor-core                                              v1.10.0                          f3a3065b3af2        2 months ago        128MB
goharbor/harbor-portal                                            v1.10.0                          fbaeb1fdacad        2 months ago        52.1MB
goharbor/harbor-db                                                v1.10.0                          634404a417cf        2 months ago        148MB
goharbor/prepare                                                  v1.10.0                          927062458494        2 months ago        149MB
openjdk                                                           8-alpine                         a3562aa0b991        9 months ago        105MB
k8s.gcr.io/kube-controller-manager                                v1.13.2                          b9027a78d94c        13 months ago       146MB
hackeruncle/kube-controller-manager                               v1.13.2                          b9027a78d94c        13 months ago       146MB
hackeruncle/kube-proxy                                            v1.13.2                          01cfa56edcfc        13 months ago       80.3MB
k8s.gcr.io/kube-proxy                                             v1.13.2                          01cfa56edcfc        13 months ago       80.3MB
hackeruncle/kube-apiserver                                        v1.13.2                          177db4b8e93a        13 months ago       181MB
k8s.gcr.io/kube-apiserver                                         v1.13.2                          177db4b8e93a        13 months ago       181MB
hackeruncle/kube-scheduler                                        v1.13.2                          3193be46e0b3        13 months ago       79.6MB
k8s.gcr.io/kube-scheduler                                         v1.13.2                          3193be46e0b3        13 months ago       79.6MB
registry.cn-beijing.aliyuncs.com/minminmsn/kubernetes-dashboard   v1.10.1                          2b631cbc40bc        13 months ago       122MB
172.21.230.89/doublehappy/hello-world                             v1                               fce289e99eb9        13 months ago       1.84kB
container01/doublehappy/hello-world                               v2                               fce289e99eb9        13 months ago       1.84kB
k8s.gcr.io/coredns                                                1.2.6                            f59dcacceff4        15 months ago       40MB
hackeruncle/coredns                                               1.2.6                            f59dcacceff4        15 months ago       40MB
hackeruncle/etcd                                                  3.2.24                           3cab8e1b9802        16 months ago       220MB
k8s.gcr.io/etcd                                                   3.2.24                           3cab8e1b9802        16 months ago       220MB
quay.io/coreos/flannel                                            v0.10.0-amd64                    f0fad859c909        2 years ago         44.6MB
hackeruncle/pause                                                 3.1                              da86e6ba6ca1        2 years ago         742kB
k8s.gcr.io/pause                                                  3.1                              da86e6ba6ca1        2 years ago         742kB
172.21.230.89/doublehappy/mysql-slave                             v1.0                             b152e49ead77        3 years ago         426MB
hackeruncle/mysql-slave                                           v1.0                             b152e49ead77        3 years ago         426MB
172.21.230.89/doublehappy/mysql-master                            v1.0                             b73ad03215d5        3 years ago         426MB
hackeruncle/mysql-master                                          v1.0                             b73ad03215d5        3 years ago         426MB
[root@container01 ~]# docker images
REPOSITORY                                                        TAG                              IMAGE ID            CREATED              SIZE
172.21.230.89/doublehappy/spark-r                                 v2.4.3                           e77ac91a3898        About a minute ago   757MB
172.21.230.89/doublehappy/spark-py                                v2.4.3                           0807413bcdcd        About a minute ago   462MB
172.21.230.89/doublehappy/spark                                   v2.4.3                           7d895d565b9d        About a minute ago   371MB
mysql                                                             doublehappy5.7                   5c68aaadf900        4 days ago           434MB
debian                                                            stretch-slim                     5174edbf8c39        9 days ago           55.3MB
goharbor/chartmuseum-photon                                       v0.9.0-v1.10.0                   c8c98d2ac22b        2 months ago         128MB
goharbor/harbor-migrator                                          v1.10.0                          9dd0c9228c79        2 months ago         362MB
goharbor/redis-photon                                             v1.10.0                          6df66e5c1ca7        2 months ago         111MB
goharbor/clair-adapter-photon                                     v1.0.1-v1.10.0                   4bdc58f374c2        2 months ago         61.6MB
goharbor/clair-photon                                             v2.1.1-v1.10.0                   435d235c5000        2 months ago         171MB
goharbor/notary-server-photon                                     v0.6.1-v1.10.0                   c7934c32e873        2 months ago         143MB
goharbor/notary-signer-photon                                     v0.6.1-v1.10.0                   bb1a762155ab        2 months ago         140MB
goharbor/harbor-registryctl                                       v1.10.0                          c550280445e6        2 months ago         104MB
goharbor/registry-photon                                          v2.7.1-patch-2819-2553-v1.10.0   2115e08fa399        2 months ago         86.5MB
goharbor/nginx-photon                                             v1.10.0                          f7ed614c3abc        2 months ago         44MB
goharbor/harbor-log                                               v1.10.0                          fb15f6772e9a        2 months ago         82.3MB
goharbor/harbor-jobservice                                        v1.10.0                          d6d4f2b125f6        2 months ago         142MB
goharbor/harbor-core                                              v1.10.0                          f3a3065b3af2        2 months ago         128MB
goharbor/harbor-portal                                            v1.10.0                          fbaeb1fdacad        2 months ago         52.1MB
goharbor/harbor-db                                                v1.10.0                          634404a417cf        2 months ago         148MB
goharbor/prepare                                                  v1.10.0                          927062458494        2 months ago         149MB
openjdk                                                           8-alpine                         a3562aa0b991        9 months ago         105MB
hackeruncle/kube-apiserver                                        v1.13.2                          177db4b8e93a        13 months ago        181MB
k8s.gcr.io/kube-apiserver                                         v1.13.2                          177db4b8e93a        13 months ago        181MB
hackeruncle/kube-proxy                                            v1.13.2                          01cfa56edcfc        13 months ago        80.3MB
k8s.gcr.io/kube-proxy                                             v1.13.2                          01cfa56edcfc        13 months ago        80.3MB
hackeruncle/kube-controller-manager                               v1.13.2                          b9027a78d94c        13 months ago        146MB
k8s.gcr.io/kube-controller-manager                                v1.13.2                          b9027a78d94c        13 months ago        146MB
hackeruncle/kube-scheduler                                        v1.13.2                          3193be46e0b3        13 months ago        79.6MB
k8s.gcr.io/kube-scheduler                                         v1.13.2                          3193be46e0b3        13 months ago        79.6MB
registry.cn-beijing.aliyuncs.com/minminmsn/kubernetes-dashboard   v1.10.1                          2b631cbc40bc        13 months ago        122MB
172.21.230.89/doublehappy/hello-world                             v1                               fce289e99eb9        13 months ago        1.84kB
container01/doublehappy/hello-world                               v2                               fce289e99eb9        13 months ago        1.84kB
k8s.gcr.io/coredns                                                1.2.6                            f59dcacceff4        15 months ago        40MB
hackeruncle/coredns                                               1.2.6                            f59dcacceff4        15 months ago        40MB
k8s.gcr.io/etcd                                                   3.2.24                           3cab8e1b9802        16 months ago        220MB
hackeruncle/etcd                                                  3.2.24                           3cab8e1b9802        16 months ago        220MB
quay.io/coreos/flannel                                            v0.10.0-amd64                    f0fad859c909        2 years ago          44.6MB
hackeruncle/pause                                                 3.1                              da86e6ba6ca1        2 years ago          742kB
k8s.gcr.io/pause                                                  3.1                              da86e6ba6ca1        2 years ago          742kB
172.21.230.89/doublehappy/mysql-slave                             v1.0                             b152e49ead77        3 years ago          426MB
hackeruncle/mysql-slave                                           v1.0                             b152e49ead77        3 years ago          426MB
172.21.230.89/doublehappy/mysql-master                            v1.0                             b73ad03215d5        3 years ago          426MB
hackeruncle/mysql-master                                          v1.0                             b73ad03215d5        3 years ago          426MB
[root@container01 ~]# 


push：到私服

[root@container01 spark]# ./bin/docker-image-tool.sh -r 172.21.230.89/doublehappy -t v2.4.3 push
The push refers to repository [172.21.230.89/doublehappy/spark]
64f8d2d5dc6b: Pushed 
8bad673cc19a: Pushed 
37a326771171: Pushed 
6334bffbbe5f: Pushed 
e47081996b74: Pushed 
897cdaf31bf7: Pushed 
261b1b5eb382: Pushed 
5add1f9f7507: Pushed 
ceaf9e1ebef5: Pushed 
9b9b7f3d56a0: Pushed 
f1b5933fe4b5: Pushed 
v2.4.3: digest: sha256:c8f97b50e7c325a9ad342afa7e796ae594f802306fae82b82d9622a462b32fa3 size: 2624
The push refers to repository [172.21.230.89/doublehappy/spark-py]
b161c5b08e39: Pushed 
abd349f5b3b6: Pushed 
e48b59a107cb: Pushed 
64f8d2d5dc6b: Mounted from doublehappy/spark 
8bad673cc19a: Mounted from doublehappy/spark 
37a326771171: Mounted from doublehappy/spark 
6334bffbbe5f: Mounted from doublehappy/spark 
e47081996b74: Mounted from doublehappy/spark 
897cdaf31bf7: Mounted from doublehappy/spark 
261b1b5eb382: Mounted from doublehappy/spark 
5add1f9f7507: Mounted from doublehappy/spark 
ceaf9e1ebef5: Mounted from doublehappy/spark 
9b9b7f3d56a0: Mounted from doublehappy/spark 
f1b5933fe4b5: Mounted from doublehappy/spark 
v2.4.3: digest: sha256:d466a3c0ac638e6c14048aabf6d48307f9c1ee691a8817ef5a40aeb24e420824 size: 3253
The push refers to repository [172.21.230.89/doublehappy/spark-r]
9f1cc9476665: Pushed 
4ef658c87c9d: Pushed 
13685a4568e4: Pushed 
64f8d2d5dc6b: Mounted from doublehappy/spark-py 
8bad673cc19a: Mounted from doublehappy/spark-py 
37a326771171: Mounted from doublehappy/spark-py 
6334bffbbe5f: Mounted from doublehappy/spark-py 
e47081996b74: Mounted from doublehappy/spark-py 
897cdaf31bf7: Mounted from doublehappy/spark-py 
261b1b5eb382: Mounted from doublehappy/spark-py 
5add1f9f7507: Mounted from doublehappy/spark-py 
ceaf9e1ebef5: Mounted from doublehappy/spark-py 
9b9b7f3d56a0: Mounted from doublehappy/spark-py 
f1b5933fe4b5: Mounted from doublehappy/spark-py 
v2.4.3: digest: sha256:9aad314aae767666e0fa6605753ab95d758b7286ef63fca6e7bab5732cbe4030 size: 3255
[root@container01 spark]#

在这里插入图片描述

3.创建secret
kubectl create secret docker-registry sparkkey \
--namespace=default \
--docker-server=172.21.230.89 \
--docker-username=admin \
--docker-password=Harbor12345 

[root@container01 spark]# kubectl create secret docker-registry sparkkey \
> --namespace=default \
> --docker-server=172.21.230.89 \
> --docker-username=admin \
> --docker-password=Harbor12345 
secret/sparkkey created
[root@container01 spark]# 

4.创建serviceAccount及 绑定clusterrolebinding
kubectl create serviceaccount spark

kubectl create clusterrolebinding spark-role --clusterrole=edit \
--serviceaccount=default:spark --namespace=default

[root@container01 spark]# kubectl create serviceaccount spark
serviceaccount/spark created
[root@container01 spark]# kubectl create clusterrolebinding spark-role --clusterrole=edit \
> --serviceaccount=default:spark --namespace=default
clusterrolebinding.rbac.authorization.k8s.io/spark-role created
[root@container01 spark]# 



5. 提交之前 查看 k8s port :k8s-apiserver-port (看spark官网)

[root@container01 spark]# kubectl cluster-info
Kubernetes master is running at https://172.21.230.89:6443
KubeDNS is running at https://172.21.230.89:6443/api/v1/namespaces/kube-system/services/kube-dns:dns/proxy

To further debug and diagnose cluster problems, use 'kubectl cluster-info dump'.
[root@container01 spark]# 

即：https://172.21.230.89:6443

bin/spark-submit \
--master k8s://https://172.21.230.89:6443 \
--deploy-mode cluster \
--name spark-pi \
--class org.apache.spark.examples.SparkPi \
--conf spark.executor.instances=2 \
--conf spark.kubernetes.container.image.pullSecrets=sparkkey \
--conf spark.kubernetes.container.image=172.21.230.89/doublehappy/spark:v2.4.3 \
--conf spark.kubernetes.authenticate.driver.serviceAccountName=spark \
--conf spark.kubernetes.driver.pod.name=spark-pi-driver \
local:///opt/spark/examples/jars/spark-examples_2.11-2.4.3.jar

注意：
这些参数spark官网都有 下面又截图

[root@container01 spark]# bin/spark-submit \
> --master k8s://https://172.21.230.89:6443 \
> --deploy-mode cluster \
> --name spark-pi \
> --class org.apache.spark.examples.SparkPi \
> --conf spark.executor.instances=2 \
> --conf spark.kubernetes.container.image.pullSecrets=sparkkey \
> --conf spark.kubernetes.container.image=172.21.230.89/doublehappy/spark:v2.4.3 \
> --conf spark.kubernetes.authenticate.driver.serviceAccountName=spark \
> --conf spark.kubernetes.driver.pod.name=spark-pi-driver \
> local:///opt/spark/examples/jars/spark-examples_2.11-2.4.3.jar
log4j:WARN No appenders could be found for logger (io.fabric8.kubernetes.client.Config).
log4j:WARN Please initialize the log4j system properly.
log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more info.
Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties
20/02/11 16:16:49 INFO LoggingPodStatusWatcherImpl: State changed, new state: 
         pod name: spark-pi-driver
         namespace: default
         labels: spark-app-selector -> spark-961842d41a564fc3ba906855f8ee4b57, spark-role -> driver
         pod uid: d9ab6302-4ca6-11ea-adad-00163e041ece
         creation time: 2020-02-11T08:16:48Z
         service account name: spark
         volumes: spark-local-dir-1, spark-conf-volume, spark-token-9hgdn
         node name: N/A
         start time: N/A
         container images: N/A
         phase: Pending
         status: []
20/02/11 16:16:49 INFO LoggingPodStatusWatcherImpl: State changed, new state: 
         pod name: spark-pi-driver
         namespace: default
         labels: spark-app-selector -> spark-961842d41a564fc3ba906855f8ee4b57, spark-role -> driver
         pod uid: d9ab6302-4ca6-11ea-adad-00163e041ece
         creation time: 2020-02-11T08:16:48Z
         service account name: spark
         volumes: spark-local-dir-1, spark-conf-volume, spark-token-9hgdn
         node name: container02
         start time: N/A
         container images: N/A
         phase: Pending
         status: []
20/02/11 16:16:49 INFO LoggingPodStatusWatcherImpl: State changed, new state: 
         pod name: spark-pi-driver
         namespace: default
         labels: spark-app-selector -> spark-961842d41a564fc3ba906855f8ee4b57, spark-role -> driver
         pod uid: d9ab6302-4ca6-11ea-adad-00163e041ece
         creation time: 2020-02-11T08:16:48Z
         service account name: spark
         volumes: spark-local-dir-1, spark-conf-volume, spark-token-9hgdn
         node name: container02
         start time: 2020-02-11T08:16:49Z
         container images: 172.21.230.89/doublehappy/spark:v2.4.3
         phase: Pending
         status: [ContainerStatus(containerID=null, image=172.21.230.89/doublehappy/spark:v2.4.3, imageID=, lastState=ContainerState(running=null, terminated=null, waiting=null, additionalProperties={}), name=spark-kubernetes-driver, ready=false, restartCount=0, state=ContainerState(running=null, terminated=null, waiting=ContainerStateWaiting(message=null, reason=ContainerCreating, additionalProperties={}), additionalProperties={}), additionalProperties={})]
20/02/11 16:16:49 INFO Client: Waiting for application spark-pi to finish...
20/02/11 16:16:59 INFO LoggingPodStatusWatcherImpl: State changed, new state: 
         pod name: spark-pi-driver
         namespace: default
         labels: spark-app-selector -> spark-961842d41a564fc3ba906855f8ee4b57, spark-role -> driver
         pod uid: d9ab6302-4ca6-11ea-adad-00163e041ece
         creation time: 2020-02-11T08:16:48Z
         service account name: spark
         volumes: spark-local-dir-1, spark-conf-volume, spark-token-9hgdn
         node name: container02
         start time: 2020-02-11T08:16:49Z
         container images: 172.21.230.89/doublehappy/spark:v2.4.3
         phase: Running
         status: [ContainerStatus(containerID=docker://80f7f9b25b92c8995b99de55ad1bcd47d46e04bbfe794f09b394a8516468ac2b, image=172.21.230.89/doublehappy/spark:v2.4.3, imageID=docker-pullable://172.21.230.89/doublehappy/spark@sha256:c8f97b50e7c325a9ad342afa7e796ae594f802306fae82b82d9622a462b32fa3, lastState=ContainerState(running=null, terminated=null, waiting=null, additionalProperties={}), name=spark-kubernetes-driver, ready=true, restartCount=0, state=ContainerState(running=ContainerStateRunning(startedAt=2020-02-11T08:16:59Z, additionalProperties={}), terminated=null, waiting=null, additionalProperties={}), additionalProperties={})]
20/02/11 16:17:34 INFO LoggingPodStatusWatcherImpl: State changed, new state: 
         pod name: spark-pi-driver
         namespace: default
         labels: spark-app-selector -> spark-961842d41a564fc3ba906855f8ee4b57, spark-role -> driver
         pod uid: d9ab6302-4ca6-11ea-adad-00163e041ece
         creation time: 2020-02-11T08:16:48Z
         service account name: spark
         volumes: spark-local-dir-1, spark-conf-volume, spark-token-9hgdn
         node name: container02
         start time: 2020-02-11T08:16:49Z
         container images: 172.21.230.89/doublehappy/spark:v2.4.3
         phase: Succeeded
         status: [ContainerStatus(containerID=docker://80f7f9b25b92c8995b99de55ad1bcd47d46e04bbfe794f09b394a8516468ac2b, image=172.21.230.89/doublehappy/spark:v2.4.3, imageID=docker-pullable://172.21.230.89/doublehappy/spark@sha256:c8f97b50e7c325a9ad342afa7e796ae594f802306fae82b82d9622a462b32fa3, lastState=ContainerState(running=null, terminated=null, waiting=null, additionalProperties={}), name=spark-kubernetes-driver, ready=false, restartCount=0, state=ContainerState(running=null, terminated=ContainerStateTerminated(containerID=docker://80f7f9b25b92c8995b99de55ad1bcd47d46e04bbfe794f09b394a8516468ac2b, exitCode=0, finishedAt=2020-02-11T08:17:34Z, message=null, reason=Completed, signal=null, startedAt=2020-02-11T08:16:59Z, additionalProperties={}), waiting=null, additionalProperties={}), additionalProperties={})]
20/02/11 16:17:34 INFO LoggingPodStatusWatcherImpl: Container final statuses:


         Container name: spark-kubernetes-driver
         Container image: 172.21.230.89/doublehappy/spark:v2.4.3
         Container state: Terminated
         Exit code: 0
20/02/11 16:17:34 INFO Client: Application spark-pi finished.
20/02/11 16:17:34 INFO ShutdownHookManager: Shutdown hook called
20/02/11 16:17:34 INFO ShutdownHookManager: Deleting directory /tmp/spark-41d1f84f-677e-4476-9cdc-a277ebe96dfa
[root@container01 spark]# 

查看一下：
[root@container01 ~]# kubectl get  all
NAME                  READY   STATUS      RESTARTS   AGE
pod/spark-pi-driver   0/1     Completed   0          92s

NAME                                        TYPE        CLUSTER-IP   EXTERNAL-IP   PORT(S)             AGE
service/kubernetes                          ClusterIP   10.96.0.1    <none>        443/TCP             26h
service/spark-pi-1581409007077-driver-svc   ClusterIP   None         <none>        7078/TCP,7079/TCP   91s
[root@container01 ~]#

查看spark log：

[root@container01 ~]# kubectl logs -f spark-pi-driver
++ id -u
+ myuid=0
++ id -g
+ mygid=0
+ set +e
++ getent passwd 0
+ uidentry=root:x:0:0:root:/root:/bin/ash
+ set -e
+ '[' -z root:x:0:0:root:/root:/bin/ash ']'
+ SPARK_K8S_CMD=driver
+ case "$SPARK_K8S_CMD" in
+ shift 1
+ SPARK_CLASSPATH=':/opt/spark/jars/*'
+ grep SPARK_JAVA_OPT_
+ sort -t_ -k4 -n
+ sed 's/[^=]*=\(.*\)/\1/g'
+ env
+ readarray -t SPARK_EXECUTOR_JAVA_OPTS
+ '[' -n '' ']'
+ '[' -n '' ']'
+ PYSPARK_ARGS=
+ '[' -n '' ']'
+ R_ARGS=
+ '[' -n '' ']'
+ '[' '' == 2 ']'
+ '[' '' == 3 ']'
+ case "$SPARK_K8S_CMD" in
+ CMD=("$SPARK_HOME/bin/spark-submit" --conf "spark.driver.bindAddress=$SPARK_DRIVER_BIND_ADDRESS" --deploy-mode client "$@")
+ exec /sbin/tini -s -- /opt/spark/bin/spark-submit --conf spark.driver.bindAddress=10.244.1.5 --deploy-mode client --properties-file /opt/spark/conf/spark.properties --class org.apache.spark.examples.SparkPi spark-internal
20/02/11 08:17:00 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties
20/02/11 08:17:01 INFO SparkContext: Running Spark version 2.4.3
20/02/11 08:17:01 INFO SparkContext: Submitted application: Spark Pi
20/02/11 08:17:01 INFO SecurityManager: Changing view acls to: root
20/02/11 08:17:01 INFO SecurityManager: Changing modify acls to: root
20/02/11 08:17:01 INFO SecurityManager: Changing view acls groups to: 
20/02/11 08:17:01 INFO SecurityManager: Changing modify acls groups to: 
20/02/11 08:17:01 INFO SecurityManager: SecurityManager: authentication disabled; ui acls disabled; users  with view permissions: Set(root); groups with view permissions: Set(); users  with modify permissions: Set(root); groups with modify permissions: Set()
20/02/11 08:17:01 INFO Utils: Successfully started service 'sparkDriver' on port 7078.
20/02/11 08:17:01 INFO SparkEnv: Registering MapOutputTracker
20/02/11 08:17:01 INFO SparkEnv: Registering BlockManagerMaster
20/02/11 08:17:01 INFO BlockManagerMasterEndpoint: Using org.apache.spark.storage.DefaultTopologyMapper for getting topology information
20/02/11 08:17:01 INFO BlockManagerMasterEndpoint: BlockManagerMasterEndpoint up
20/02/11 08:17:01 INFO DiskBlockManager: Created local directory at /var/data/spark-6753da05-ab60-4e01-bafd-70e0141b47f2/blockmgr-f9c8a8f5-32c4-42b4-bb3b-3ec9686c04f5
20/02/11 08:17:01 INFO MemoryStore: MemoryStore started with capacity 413.9 MB
20/02/11 08:17:01 INFO SparkEnv: Registering OutputCommitCoordinator
20/02/11 08:17:01 INFO Utils: Successfully started service 'SparkUI' on port 4040.
20/02/11 08:17:01 INFO SparkUI: Bound SparkUI to 0.0.0.0, and started at http://spark-pi-1581409007077-driver-svc.default.svc:4040
20/02/11 08:17:01 INFO SparkContext: Added JAR file:///opt/spark/examples/jars/spark-examples_2.11-2.4.3.jar at spark://spark-pi-1581409007077-driver-svc.default.svc:7078/jars/spark-examples_2.11-2.4.3.jar with timestamp 1581409021888
20/02/11 08:17:03 INFO ExecutorPodsAllocator: Going to request 2 executors from Kubernetes.
20/02/11 08:17:03 INFO Utils: Successfully started service 'org.apache.spark.network.netty.NettyBlockTransferService' on port 7079.
20/02/11 08:17:03 INFO NettyBlockTransferService: Server created on spark-pi-1581409007077-driver-svc.default.svc:7079
20/02/11 08:17:03 INFO BlockManager: Using org.apache.spark.storage.RandomBlockReplicationPolicy for block replication policy
20/02/11 08:17:03 INFO BlockManagerMaster: Registering BlockManager BlockManagerId(driver, spark-pi-1581409007077-driver-svc.default.svc, 7079, None)
20/02/11 08:17:03 INFO BlockManagerMasterEndpoint: Registering block manager spark-pi-1581409007077-driver-svc.default.svc:7079 with 413.9 MB RAM, BlockManagerId(driver, spark-pi-1581409007077-driver-svc.default.svc, 7079, None)
20/02/11 08:17:03 INFO BlockManagerMaster: Registered BlockManager BlockManagerId(driver, spark-pi-1581409007077-driver-svc.default.svc, 7079, None)
20/02/11 08:17:03 INFO BlockManager: Initialized BlockManager: BlockManagerId(driver, spark-pi-1581409007077-driver-svc.default.svc, 7079, None)
20/02/11 08:17:14 INFO KubernetesClusterSchedulerBackend$KubernetesDriverEndpoint: Registered executor NettyRpcEndpointRef(spark-client://Executor) (10.244.2.4:56814) with ID 1
20/02/11 08:17:14 INFO BlockManagerMasterEndpoint: Registering block manager 10.244.2.4:37098 with 413.9 MB RAM, BlockManagerId(1, 10.244.2.4, 37098, None)
20/02/11 08:17:33 INFO KubernetesClusterSchedulerBackend: SchedulerBackend is ready for scheduling beginning after waiting maxRegisteredResourcesWaitingTime: 30000(ms)
20/02/11 08:17:33 INFO SparkContext: Starting job: reduce at SparkPi.scala:38
20/02/11 08:17:33 INFO DAGScheduler: Got job 0 (reduce at SparkPi.scala:38) with 2 output partitions
20/02/11 08:17:33 INFO DAGScheduler: Final stage: ResultStage 0 (reduce at SparkPi.scala:38)
20/02/11 08:17:33 INFO DAGScheduler: Parents of final stage: List()
20/02/11 08:17:33 INFO DAGScheduler: Missing parents: List()
20/02/11 08:17:33 INFO DAGScheduler: Submitting ResultStage 0 (MapPartitionsRDD[1] at map at SparkPi.scala:34), which has no missing parents
20/02/11 08:17:33 INFO MemoryStore: Block broadcast_0 stored as values in memory (estimated size 1936.0 B, free 413.9 MB)
20/02/11 08:17:33 INFO MemoryStore: Block broadcast_0_piece0 stored as bytes in memory (estimated size 1256.0 B, free 413.9 MB)
20/02/11 08:17:33 INFO BlockManagerInfo: Added broadcast_0_piece0 in memory on spark-pi-1581409007077-driver-svc.default.svc:7079 (size: 1256.0 B, free: 413.9 MB)
20/02/11 08:17:33 INFO SparkContext: Created broadcast 0 from broadcast at DAGScheduler.scala:1161
20/02/11 08:17:33 INFO DAGScheduler: Submitting 2 missing tasks from ResultStage 0 (MapPartitionsRDD[1] at map at SparkPi.scala:34) (first 15 tasks are for partitions Vector(0, 1))
20/02/11 08:17:33 INFO TaskSchedulerImpl: Adding task set 0.0 with 2 tasks
20/02/11 08:17:33 INFO TaskSetManager: Starting task 0.0 in stage 0.0 (TID 0, 10.244.2.4, executor 1, partition 0, PROCESS_LOCAL, 7885 bytes)
20/02/11 08:17:34 INFO BlockManagerInfo: Added broadcast_0_piece0 in memory on 10.244.2.4:37098 (size: 1256.0 B, free: 413.9 MB)
20/02/11 08:17:34 INFO TaskSetManager: Starting task 1.0 in stage 0.0 (TID 1, 10.244.2.4, executor 1, partition 1, PROCESS_LOCAL, 7885 bytes)
20/02/11 08:17:34 INFO TaskSetManager: Finished task 0.0 in stage 0.0 (TID 0) in 534 ms on 10.244.2.4 (executor 1) (1/2)
20/02/11 08:17:34 INFO TaskSetManager: Finished task 1.0 in stage 0.0 (TID 1) in 27 ms on 10.244.2.4 (executor 1) (2/2)
20/02/11 08:17:34 INFO TaskSchedulerImpl: Removed TaskSet 0.0, whose tasks have all completed, from pool 
20/02/11 08:17:34 INFO DAGScheduler: ResultStage 0 (reduce at SparkPi.scala:38) finished in 0.814 s
20/02/11 08:17:34 INFO DAGScheduler: Job 0 finished: reduce at SparkPi.scala:38, took 0.868531 s
Pi is roughly 3.139475697378487
20/02/11 08:17:34 INFO SparkUI: Stopped Spark web UI at http://spark-pi-1581409007077-driver-svc.default.svc:4040
20/02/11 08:17:34 INFO KubernetesClusterSchedulerBackend: Shutting down all executors
20/02/11 08:17:34 INFO KubernetesClusterSchedulerBackend$KubernetesDriverEndpoint: Asking each executor to shut down
20/02/11 08:17:34 WARN ExecutorPodsWatchSnapshotSource: Kubernetes client has been closed (this is expected if the application is shutting down.)
20/02/11 08:17:34 INFO MapOutputTrackerMasterEndpoint: MapOutputTrackerMasterEndpoint stopped!
20/02/11 08:17:34 INFO MemoryStore: MemoryStore cleared
20/02/11 08:17:34 INFO BlockManager: BlockManager stopped
20/02/11 08:17:34 INFO BlockManagerMaster: BlockManagerMaster stopped
20/02/11 08:17:34 INFO OutputCommitCoordinator$OutputCommitCoordinatorEndpoint: OutputCommitCoordinator stopped!
20/02/11 08:17:34 INFO SparkContext: Successfully stopped SparkContext
20/02/11 08:17:34 INFO ShutdownHookManager: Shutdown hook called
20/02/11 08:17:34 INFO ShutdownHookManager: Deleting directory /var/data/spark-6753da05-ab60-4e01-bafd-70e0141b47f2/spark-67555fc1-109a-46a0-bff7-1c14b972cfe4
20/02/11 08:17:34 INFO ShutdownHookManager: Deleting directory /tmp/spark-7f09677c-5829-434e-b74d-af0f0c021571
[root@container01 ~]# 

结果：使用kubectl logs -f spark-pi-driver 是可以看到的

Pi is roughly 3.1451757258786293
20/02/11 08:23:35 INFO SparkUI: Stopped Spark web UI at http://spark-pi-1581409376163-driver-svc.default.svc:4040
20/02/11 08:23:35 INFO KubernetesClusterSchedulerBacke


注意：
1.上面的spark命令在 spark官网的基础上添加了 很多参数  官网上也能找到
2.机器上要 先部署 jdk

如果没有加 ：
spark 出错了 使用 
	1.kubectl get all   先看看有没有error
	2.再查看spark （pod name） 通过  
	  kubectl logs -f spark-pi-driver  
	3.如果出问题了 先 delete
		kubectl delete [pod name] 
		[root@container01 ~]# kubectl  delete pod/spark-pi-driver 
			pod "spark-pi-driver" deleted
		[root@container01 ~]# 		
   4.再排错 重新提交

在这里插入图片描述

Kudu+Impala故障案例01--double_happy

2020-01-05

背景

Flink消费数据 最终写入Kudu里面进行存储，
impala通过创建外部表的方式，关联kudu里的数据 进行后续的指标计算

注意：
   1.Flink ->Sink kudu 
   2.Impala ->关联kudu

故障

1.Flink sink kudu  表的名字 
 eg： kudu01.test01  这个表 
 Impala关联 kudu 之后 impala里的表是 kudu_test01

即：
  kudu01.test01 -->  kudu_test01

故障：
	当把kudu01.test01 表删掉  (通过kudu api 删掉表)  
	重新创建的 kudu01.test01 	
删掉kudu表直接重建 但是 impala没有重新关联 会造成 出现 null

[vm24:21000] > select * from kudu_access08 limit 100;
Query: select * from kudu_access08 limit 100
Query submitted at: 2019-12-31 11:31:55 (Coordinator: http://vm24:25000)
Query progress can be monitored at: http://vm24:25000/query_plan?query_id=f24990121ccc11cb:91d30b3d00000000
+-------------+---------------+---------+---------------+
| id          | domain        | traffic | time          |
+-------------+---------------+---------+---------------+
| 0_0_3_4_4_9 | NULL          | NULL    | NULL          |
| 0_0_3_9_7_7 | NULL          | NULL    | 1577763046142 |
| 0_0_5_1_2_3 | zhibo8.cc     | 632     | NULL          |
| 0_0_8_3_3_0 | zhibo8.cc     | 349     | 1577763045141 |
| 0_0_8_8_0_2 | NULL          | NULL    | 1577763046142 |
| 0_1_5_8_6_8 | NULL          | 256     | NULL          |
| 0_1_9_0_6_1 | ruozedata.com | 513     | 1577763045141 |
| 0_1_9_3_7_8 | dongqiudi.com | NULL    | NULL          |
| 0_2_0_0_2_1 | NULL          | 1212    | NULL          |
| 0_2_0_7_7_8 | NULL          | 1323    | 1577763045141 |
| 0_2_3_1_3_9 | ruozedata.com | NULL    | NULL          |
| 0_2_6_2_9_3 | ruozedata.com | NULL    | 1577763043139 |
| 0_2_6_8_8_9 | NULL          | 956     | 1577763046142 |
| 0_2_7_2_2_7 | NULL          | NULL    | NULL          |
| 0_2_7_7_4_2 | zhibo8.cc     | NULL    | 1577763046142 |
| 0_3_2_3_2_2 | dongqiudi.com | 309     | NULL          |
| 0_3_2_6_5_6 | NULL          | NULL    | NULL          |
| 0_3_8_9_8_0 | NULL          | NULL    | 1577763043139 |


删掉kudu表直接重建 但是 impala没有重新关联 会造成 出现 null

目前解决办法就是 ：
	通过impala把 impala关联的kudu表删掉  + 再把kudu里的表删掉 两个都重新创建 

因为kudu+impala 刚刚接触 慢慢摸索当中

Flink04--double_happy

2020-01-05

面试题：谈谈你对Flink ON YARN执行流程的理解

acks=all 数据一定不会丢失吗
如何保证Kafka消费者数据全局有序？ 伪命题
    producer
    storage
    consumer



10分区 3个消费者
    哪个分区被谁消费

C0
C1

T0
    0 1 2
T1
    0 1 2

6/2
We then divide the number of partitions by the total number of
consumers to determine the number of partitions to assign to each consumer. If it does not evenly
divide, then the first few consumers will have one extra partition.




t0p0, t1p0, t1p1, t2p0, t2p1, t2p2

C0 is subscribed to t0;           t0p0
C1 is subscribed to t0, t1;       t0p0, t1p0, t1p1
C2 is subscribed to t0, t1, t2.   t0p0, t1p0, t1p1, t2p0, t2p1, t2p2

Sinl 到 HBase：
	1种.自定义 RichSinkFunction  或者 带并行的
	2种 .  Hadoop Compatibility Beta
			You can:
			use Hadoop’s Writable data types in Flink programs.
			use any Hadoop InputFormat as a DataSource.
			use any Hadoop OutputFormat as a DataSink.
			use a Hadoop Mapper as FlatMapFunction.
			use a Hadoop Reducer as GroupReduceFunction.

Hadoop Compatibility Beta
Hadoop Compatibility Beta

在这里插入图片描述

也就是说  ：
	写入HBase 有两种方式
		1.自定义 Sink
		2.Flink-Hadoop 的Api

环境部署
Building Flink from Source

部署：
在这里插入图片描述

部署：
	1.本地的 
	2.yarn
作为重点。

Local
Local Setup Tutorial

1.启动集群：

[double_happy@hadoop101 flink]$ bin/start-cluster.sh 
Starting cluster.
Starting standalonesession daemon on host hadoop101.
Starting taskexecutor daemon on host hadoop101.

注意：
	1.Starting standalonesession 
	2.Starting taskexecutor


[double_happy@hadoop101 flink]$ jps
4688 TaskManagerRunner
4722 Jps
4258 StandaloneSessionClusterEntrypoint
[double_happy@hadoop101 flink]$ 


2.看页面 端口 8081

在这里插入图片描述

按照官网给的案例：

object SocketWindowWordCount {
  def main(args: Array[String]) : Unit = {
    val port: Int = try {
      ParameterTool.fromArgs(args).getInt("port")
    } catch {
      case e: Exception => {
        System.err.println("No port specified. Please run 'SocketWindowWordCount --host <host> --port <port>'")
        return
      }
    }

    val host: String = try {
      ParameterTool.fromArgs(args).get("host")
    } catch {
      case e: Exception => {
        System.err.println("No port specified. Please run 'SocketWindowWordCount --host <host> --port <port>'")
        return
      }
    }
    

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    val text = env.socketTextStream(host, port)
    val windowCounts = text
      .flatMap { w => w.split(",") }
      .map { w => WordWithCount(w, 1) }
      .keyBy("word")
      .sum("count")

    windowCounts.print().setParallelism(1)

    env.execute("Socket Window WordCount")
  }

  case class WordWithCount(word: String, count: Long)

}


注意：
	打包上传到本地

启动 端口 nc ：nc -lk 9998

提交flink jar 包

[double_happy@hadoop101 bin]$ ./flink
2020-01-05 23:20:00,751 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                 - Found Yarn properties file under /tmp/.yarn-properties-double_happy.
2020-01-05 23:20:00,751 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                 - Found Yarn properties file under /tmp/.yarn-properties-double_happy.
./flink <ACTION> [OPTIONS] [ARGUMENTS]

The following actions are available:

Action "run" compiles and runs a program.

  Syntax: run [OPTIONS] <jar-file> <arguments>
  "run" action options:
     -c,--class <classname>               Class with the program entry point
                                          ("main()" method or "getPlan()"
                                          method). Only needed if the JAR file
                                          does not specify the class in its
                                          manifest.
     -C,--classpath <url>                 Adds a URL to each user code
                                          classloader  on all nodes in the
                                          cluster. The paths must specify a
                                          protocol (e.g. file://) and be
                                          accessible on all nodes (e.g. by means
                                          of a NFS share). You can use this
                                          option multiple times for specifying
                                          more than one URL. The protocol must
                                          be supported by the {@link
                                          java.net.URLClassLoader}.
     -d,--detached                        If present, runs the job in detached
                                          mode
     -n,--allowNonRestoredState           Allow to skip savepoint state that
                                          cannot be restored. You need to allow
                                          this if you removed an operator from
                                          your program that was part of the
                                          program when the savepoint was
                                          triggered.
     -p,--parallelism <parallelism>       The parallelism with which to run the
                                          program. Optional flag to override the
                                          default value specified in the
                                          configuration.
     -py,--python <python>                Python script with the program entry
                                          point. The dependent resources can be
                                          configured with the `--pyFiles`
                                          option.
     -pyfs,--pyFiles <pyFiles>            Attach custom python files for job.
                                          Comma can be used as the separator to
                                          specify multiple files. The standard
                                          python resource file suffixes such as
                                          .py/.egg/.zip are all supported.(eg:
                                          --pyFiles
                                          file:///tmp/myresource.zip,hdfs:///$na
                                          menode_address/myresource2.zip)
     -pym,--pyModule <pyModule>           Python module with the program entry
                                          point. This option must be used in
                                          conjunction with `--pyFiles`.
     -q,--sysoutLogging                   If present, suppress logging output to
                                          standard out.
     -s,--fromSavepoint <savepointPath>   Path to a savepoint to restore the job
                                          from (for example
                                          hdfs:///flink/savepoint-1537).
     -sae,--shutdownOnAttachedExit        If the job is submitted in attached
                                          mode, perform a best-effort cluster
                                          shutdown when the CLI is terminated
                                          abruptly, e.g., in response to a user
                                          interrupt, such as typing Ctrl + C.
  Options for yarn-cluster mode:
     -d,--detached                        If present, runs the job in detached
                                          mode
     -m,--jobmanager <arg>                Address of the JobManager (master) to
                                          which to connect. Use this flag to
                                          connect to a different JobManager than
                                          the one specified in the
                                          configuration.
     -sae,--shutdownOnAttachedExit        If the job is submitted in attached
                                          mode, perform a best-effort cluster
                                          shutdown when the CLI is terminated
                                          abruptly, e.g., in response to a user
                                          interrupt, such as typing Ctrl + C.
     -yat,--yarnapplicationType <arg>     Set a custom application type for the
                                          application on YARN
     -yD <property=value>                 use value for given property
     -yd,--yarndetached                   If present, runs the job in detached
                                          mode (deprecated; use non-YARN
                                          specific option instead)
     -yh,--yarnhelp                       Help for the Yarn session CLI.
     -yid,--yarnapplicationId <arg>       Attach to running YARN session
     -yj,--yarnjar <arg>                  Path to Flink jar file
     -yjm,--yarnjobManagerMemory <arg>    Memory for JobManager Container with
                                          optional unit (default: MB)
     -yn,--yarncontainer <arg>            Number of YARN container to allocate
                                          (=Number of Task Managers)
     -ynl,--yarnnodeLabel <arg>           Specify YARN node label for the YARN
                                          application
     -ynm,--yarnname <arg>                Set a custom name for the application
                                          on YARN
     -yq,--yarnquery                      Display available YARN resources
                                          (memory, cores)
     -yqu,--yarnqueue <arg>               Specify YARN queue.
     -ys,--yarnslots <arg>                Number of slots per TaskManager
     -yst,--yarnstreaming                 Start Flink in streaming mode
     -yt,--yarnship <arg>                 Ship files in the specified directory
                                          (t for transfer)
     -ytm,--yarntaskManagerMemory <arg>   Memory per TaskManager Container with
                                          optional unit (default: MB)
     -yz,--yarnzookeeperNamespace <arg>   Namespace to create the Zookeeper
                                          sub-paths for high availability mode
     -z,--zookeeperNamespace <arg>        Namespace to create the Zookeeper
                                          sub-paths for high availability mode

  Options for default mode:
     -m,--jobmanager <arg>           Address of the JobManager (master) to which
                                     to connect. Use this flag to connect to a
                                     different JobManager than the one specified
                                     in the configuration.
     -z,--zookeeperNamespace <arg>   Namespace to create the Zookeeper sub-paths
                                     for high availability mode



Action "info" shows the optimized execution plan of the program (JSON).

  Syntax: info [OPTIONS] <jar-file> <arguments>
  "info" action options:
     -c,--class <classname>           Class with the program entry point
                                      ("main()" method or "getPlan()" method).
                                      Only needed if the JAR file does not
                                      specify the class in its manifest.
     -p,--parallelism <parallelism>   The parallelism with which to run the
                                      program. Optional flag to override the
                                      default value specified in the
                                      configuration.


Action "list" lists running and scheduled programs.

  Syntax: list [OPTIONS]
  "list" action options:
     -r,--running     Show only running programs and their JobIDs
     -s,--scheduled   Show only scheduled programs and their JobIDs
  Options for yarn-cluster mode:
     -m,--jobmanager <arg>            Address of the JobManager (master) to
                                      which to connect. Use this flag to connect
                                      to a different JobManager than the one
                                      specified in the configuration.
     -yid,--yarnapplicationId <arg>   Attach to running YARN session
     -z,--zookeeperNamespace <arg>    Namespace to create the Zookeeper
                                      sub-paths for high availability mode

  Options for default mode:
     -m,--jobmanager <arg>           Address of the JobManager (master) to which
                                     to connect. Use this flag to connect to a
                                     different JobManager than the one specified
                                     in the configuration.
     -z,--zookeeperNamespace <arg>   Namespace to create the Zookeeper sub-paths
                                     for high availability mode



Action "stop" stops a running program with a savepoint (streaming jobs only).

  Syntax: stop [OPTIONS] <Job ID>
  "stop" action options:
     -d,--drain                           Send MAX_WATERMARK before taking the
                                          savepoint and stopping the pipelne.
     -p,--savepointPath <savepointPath>   Path to the savepoint (for example
                                          hdfs:///flink/savepoint-1537). If no
                                          directory is specified, the configured
                                          default will be used
                                          ("state.savepoints.dir").
  Options for yarn-cluster mode:
     -m,--jobmanager <arg>            Address of the JobManager (master) to
                                      which to connect. Use this flag to connect
                                      to a different JobManager than the one
                                      specified in the configuration.
     -yid,--yarnapplicationId <arg>   Attach to running YARN session
     -z,--zookeeperNamespace <arg>    Namespace to create the Zookeeper
                                      sub-paths for high availability mode

  Options for default mode:
     -m,--jobmanager <arg>           Address of the JobManager (master) to which
                                     to connect. Use this flag to connect to a
                                     different JobManager than the one specified
                                     in the configuration.
     -z,--zookeeperNamespace <arg>   Namespace to create the Zookeeper sub-paths
                                     for high availability mode



Action "cancel" cancels a running program.

  Syntax: cancel [OPTIONS] <Job ID>
  "cancel" action options:
     -s,--withSavepoint <targetDirectory>   **DEPRECATION WARNING**: Cancelling
                                            a job with savepoint is deprecated.
                                            Use "stop" instead.
                                            Trigger savepoint and cancel job.
                                            The target directory is optional. If
                                            no directory is specified, the
                                            configured default directory
                                            (state.savepoints.dir) is used.
  Options for yarn-cluster mode:
     -m,--jobmanager <arg>            Address of the JobManager (master) to
                                      which to connect. Use this flag to connect
                                      to a different JobManager than the one
                                      specified in the configuration.
     -yid,--yarnapplicationId <arg>   Attach to running YARN session
     -z,--zookeeperNamespace <arg>    Namespace to create the Zookeeper
                                      sub-paths for high availability mode

  Options for default mode:
     -m,--jobmanager <arg>           Address of the JobManager (master) to which
                                     to connect. Use this flag to connect to a
                                     different JobManager than the one specified
                                     in the configuration.
     -z,--zookeeperNamespace <arg>   Namespace to create the Zookeeper sub-paths
                                     for high availability mode



Action "savepoint" triggers savepoints for a running job or disposes existing ones.

  Syntax: savepoint [OPTIONS] <Job ID> [<target directory>]
  "savepoint" action options:
     -d,--dispose <arg>       Path of savepoint to dispose.
     -j,--jarfile <jarfile>   Flink program JAR file.
  Options for yarn-cluster mode:
     -m,--jobmanager <arg>            Address of the JobManager (master) to
                                      which to connect. Use this flag to connect
                                      to a different JobManager than the one
                                      specified in the configuration.
     -yid,--yarnapplicationId <arg>   Attach to running YARN session
     -z,--zookeeperNamespace <arg>    Namespace to create the Zookeeper
                                      sub-paths for high availability mode

  Options for default mode:
     -m,--jobmanager <arg>           Address of the JobManager (master) to which
                                     to connect. Use this flag to connect to a
                                     different JobManager than the one specified
                                     in the configuration.
     -z,--zookeeperNamespace <arg>   Namespace to create the Zookeeper sub-paths
                                     for high availability mode



Please specify an action.
[double_happy@hadoop101 bin]$

./flink  run [OPTIONS] <jar-file> <arguments> 

注意：  提交jar  运行 命令

./flink run \
--class com.sx.flink04.SocketWindowWordCount \
/home/double_happy/lib/Flink-1.0.jar \
--port 9998 \
--host hadoop101

提交结果：
[double_happy@hadoop101 bin]$ ./flink run \
> --class com.sx.flink04.SocketWindowWordCount \
> /home/double_happy/lib/Flink-1.0.jar \
> --port 9998 \
> --host hadoop101
Starting execution of program

在这里插入图片描述
点进去

nc 生产一些数据：

[double_happy@hadoop101 ~]$ nc -lk 9998
a,a,a,a,bb,c


那么这些数据在Flink 哪里呢？

点击 Task Managers： 能看到很多参数的 ***
	1.能看到 jvm的 参数 
	2.Network 相关的

在这里插入图片描述

这些日志哪里来的？
	就是从文件系统读进来的

在这里插入图片描述

结果：
	1.通过 页面可以看的到的 （就是从本地结果 传到页面上的 ）
	2.通过本地也能看到

本地查看：
	[double_happy@hadoop101 log]$ tail -200f flink-double_happy-taskexecutor-1-hadoop101.out
	WordWithCount(a,1)
	WordWithCount(a,2)
	WordWithCount(a,3)
	WordWithCount(a,4)
	WordWithCount(bb,1)
	WordWithCount(c,1)


注意：
	$ tail -f log/flink-*-taskexecutor-*.out  
就可以看到本地 结果

如何停止这个作业：
	1.页面 上 running jos 点进去 
	2.本地命令

在这里插入图片描述

本地命令：
	 ./flink  cancel [OPTIONS] <Job ID>

[double_happy@hadoop101 bin]$ ./flink cancel 710a899ba9589d6a688c14931f7ef6cd
Cancelling job 710a899ba9589d6a688c14931f7ef6cd.
Cancelled job 710a899ba9589d6a688c14931f7ef6cd.
[double_happy@hadoop101 bin]$

在这里插入图片描述

关闭集群：


[double_happy@hadoop101 bin]$ jps
4688 TaskManagerRunner
4258 StandaloneSessionClusterEntrypoint
9444 Jps

[double_happy@hadoop101 bin]$ ./stop-cluster.sh 
Stopping taskexecutor daemon (pid: 4688) on host hadoop101.
Stopping standalonesession daemon (pid: 4258) on host hadoop101.

[double_happy@hadoop101 bin]$ jps
10457 Jps
[double_happy@hadoop101 bin]$

注意：
	页面也就打不开了

以上：
	就是最简单的local模式 运行jar  job

对比Spark-shell

start-scala-shell.sh

[double_happy@hadoop101 bin]$ ./start-scala-shell.sh 
Starting Flink Shell:
log4j:WARN No appenders could be found for logger (org.apache.flink.configuration.GlobalConfiguration).
log4j:WARN Please initialize the log4j system properly.
log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more info.
Error: please specify execution mode:
[local | remote <host> <port> | yarn]
[double_happy@hadoop101 bin]$ 

注意：
	Error: please specify execution mode: 指定跑在什么模式下 
	Spark和Flink都可以跑在多个模式上的

[double_happy@hadoop101 bin]$ ./start-scala-shell.sh local
Starting Flink Shell:

Starting local Flink cluster (host: localhost, port: 8081).


Connecting to Flink cluster (host: localhost, port: 8081).


                         ▒▓██▓██▒
                     ▓████▒▒█▓▒▓███▓▒
                  ▓███▓░░        ▒▒▒▓██▒  ▒
                ░██▒   ▒▒▓▓█▓▓▒░      ▒████
                ██▒         ░▒▓███▒    ▒█▒█▒
                  ░▓█            ███   ▓░▒██
                    ▓█       ▒▒▒▒▒▓██▓░▒░▓▓█
                  █░ █   ▒▒░       ███▓▓█ ▒█▒▒▒
                  ████░   ▒▓█▓      ██▒▒▒ ▓███▒
               ░▒█▓▓██       ▓█▒    ▓█▒▓██▓ ░█░
         ▓░▒▓████▒ ██         ▒█    █▓░▒█▒░▒█▒
        ███▓░██▓  ▓█           █   █▓ ▒▓█▓▓█▒
      ░██▓  ░█░            █  █▒ ▒█████▓▒ ██▓░▒
     ███░ ░ █░          ▓ ░█ █████▒░░    ░█░▓  ▓░
    ██▓█ ▒▒▓▒          ▓███████▓░       ▒█▒ ▒▓ ▓██▓
 ▒██▓ ▓█ █▓█       ░▒█████▓▓▒░         ██▒▒  █ ▒  ▓█▒
 ▓█▓  ▓█ ██▓ ░▓▓▓▓▓▓▓▒              ▒██▓           ░█▒
 ▓█    █ ▓███▓▒░              ░▓▓▓███▓          ░▒░ ▓█
 ██▓    ██▒    ░▒▓▓███▓▓▓▓▓██████▓▒            ▓███  █
▓███▒ ███   ░▓▓▒░░   ░▓████▓░                  ░▒▓▒  █▓
█▓▒▒▓▓██  ░▒▒░░░▒▒▒▒▓██▓░                            █▓
██ ▓░▒█   ▓▓▓▓▒░░  ▒█▓       ▒▓▓██▓    ▓▒          ▒▒▓
▓█▓ ▓▒█  █▓░  ░▒▓▓██▒            ░▓█▒   ▒▒▒░▒▒▓█████▒
 ██░ ▓█▒█▒  ▒▓▓▒  ▓█                █░      ░░░░   ░█▒
 ▓█   ▒█▓   ░     █░                ▒█              █▓
  █▓   ██         █░                 ▓▓        ▒█▓▓▓▒█░
   █▓ ░▓██░       ▓▒                  ▓█▓▒░░░▒▓█░    ▒█
    ██   ▓█▓░      ▒                    ░▒█▒██▒      ▓▓
     ▓█▒   ▒█▓▒░                         ▒▒ █▒█▓▒▒░░▒██
      ░██▒    ▒▓▓▒                     ▓██▓▒█▒ ░▓▓▓▓▒█▓
        ░▓██▒                          ▓░  ▒█▓█  ░░▒▒▒
            ▒▓▓▓▓▓▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒░░▓▓  ▓░▒█░

              F L I N K - S C A L A - S H E L L

NOTE: Use the prebound Execution Environments and Table Environment to implement batch or streaming programs.

  Batch - Use the 'benv' and 'btenv' variable

    * val dataSet = benv.readTextFile("/path/to/data")
    * dataSet.writeAsText("/path/to/output")
    * benv.execute("My batch program")
    *
    * val batchTable = btenv.fromDataSet(dataSet)
    * btenv.registerTable("tableName", batchTable)
    * val result = btenv.sqlQuery("SELECT * FROM tableName").collect
    HINT: You can use print() on a DataSet to print the contents or collect()
    a sql query result back to the shell.

  Streaming - Use the 'senv' and 'stenv' variable

    * val dataStream = senv.fromElements(1, 2, 3, 4)
    * dataStream.countWindowAll(2).sum(0).print()
    *
    * val streamTable = stenv.fromDataStream(dataStream, 'num)
    * val resultTable = streamTable.select('num).where('num % 2 === 1 )
    * resultTable.toAppendStream[Row].print()
    * senv.execute("My streaming program")
    HINT: You can only print a DataStream to the shell in local mode.
      

scala>

注意：
	  Batch - Use the 'benv' and 'btenv' variable
	  Streaming - Use the 'senv' and 'stenv' variable

批处理：
scala> val text = benv.fromElements("kairis,kairis,kairis","double_happy,sxwang")
text: org.apache.flink.api.scala.DataSet[String] = org.apache.flink.api.scala.DataSet@26a004ed

scala> text.flatMap(_.split(",")).map((_,1)).groupBy(0).sum(1).print
(double_happy,1)
(kairis,3)
(sxwang,1)

scala>

跑在Yarn上 ***

面试题：谈谈你对Flink ON YARN执行流程的理解？

	和Spark mr 都差不多 只是 ApplicaitonMaster 名字变了

YARN Setup
在这里插入图片描述
 Background / Internals

The YARN client needs to access the Hadoop configuration to connect to the YARN resource manager and HDFS

注意：
	YARN client 要配置gateway

但是：
	Flink跑yarn 跟Spark 跑yarn 有点区别

为什么呢？
	Flink有两种模式跑yarn
		1.Start a long-running Flink cluster on YARN
			启动一个长服务
		2.Run a Flink job on YARN

在这里插入图片描述

Spark里面 就是 右边这种  

注意：
	Spark申请资源是耗时的 还记得吗 依赖的 jar包需要传到hdfs上 也不然每次都要传 还记得么

这两种方式会带来什么好处：
	1.长服务 
			省资源申请的时间

两种方式分别演示

 ./yarn-session.sh  运行的时候有坑的 ：
 	1.你自己源码编译好的是没有问题的
 	2.不是编译的 要在flink lib包下面 加一个 lib哦


[double_happy@hadoop101 bin]$ ./yarn-session.sh --help
2020-01-05 22:32:14,068 INFO  org.apache.flink.configuration.GlobalConfiguration            - Loading configuration property: jobmanager.rpc.address, hadoop101
2020-01-05 22:32:14,070 INFO  org.apache.flink.configuration.GlobalConfiguration            - Loading configuration property: jobmanager.rpc.port, 6123
2020-01-05 22:32:14,070 INFO  org.apache.flink.configuration.GlobalConfiguration            - Loading configuration property: jobmanager.heap.size, 1024m
2020-01-05 22:32:14,070 INFO  org.apache.flink.configuration.GlobalConfiguration            - Loading configuration property: taskmanager.heap.size, 1024m
2020-01-05 22:32:14,070 INFO  org.apache.flink.configuration.GlobalConfiguration            - Loading configuration property: taskmanager.numberOfTaskSlots, 1
2020-01-05 22:32:14,070 INFO  org.apache.flink.configuration.GlobalConfiguration            - Loading configuration property: parallelism.default, 1
2020-01-05 22:32:14,071 INFO  org.apache.flink.configuration.GlobalConfiguration            - Loading configuration property: jobmanager.execution.failover-strategy, region
2020-01-05 22:32:14,071 INFO  org.apache.flink.configuration.GlobalConfiguration            - Loading configuration property: rest.port, 8081
2020-01-05 22:32:14,667 WARN  org.apache.hadoop.util.NativeCodeLoader                       - Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
2020-01-05 22:32:14,923 INFO  org.apache.flink.runtime.security.modules.HadoopModule        - Hadoop user set to double_happy (auth:SIMPLE)
Usage:
   Required
     -n,--container <arg>   Number of YARN container to allocate (=Number of Task Managers)
   Optional
     -at,--applicationType <arg>     Set a custom application type for the application on YARN
     -D <property=value>             use value for given property
     -d,--detached                   If present, runs the job in detached mode
     -h,--help                       Help for the Yarn session CLI.
     -id,--applicationId <arg>       Attach to running YARN session
     -j,--jar <arg>                  Path to Flink jar file
     -jm,--jobManagerMemory <arg>    Memory for JobManager Container with optional unit (default: MB)
     -m,--jobmanager <arg>           Address of the JobManager (master) to which to connect. Use this flag to connect to a different JobManager than the one specified in the configuration.
     -n,--container <arg>            Number of YARN container to allocate (=Number of Task Managers)
     -nl,--nodeLabel <arg>           Specify YARN node label for the YARN application
     -nm,--name <arg>                Set a custom name for the application on YARN
     -q,--query                      Display available YARN resources (memory, cores)
     -qu,--queue <arg>               Specify YARN queue.
     -s,--slots <arg>                Number of slots per TaskManager
     -sae,--shutdownOnAttachedExit   If the job is submitted in attached mode, perform a best-effort cluster shutdown when the CLI is terminated abruptly, e.g., in response to a user interrupt, such
                                     as typing Ctrl + C.
     -st,--streaming                 Start Flink in streaming mode
     -t,--ship <arg>                 Ship files in the specified directory (t for transfer)
     -tm,--taskManagerMemory <arg>   Memory per TaskManager Container with optional unit (default: MB)
     -yd,--yarndetached              If present, runs the job in detached mode (deprecated; use non-YARN specific option instead)
     -z,--zookeeperNamespace <arg>   Namespace to create the Zookeeper sub-paths for high availability mode
[double_happy@hadoop101 bin]$

  Required
     -n,--container <arg>   Number of YARN container to allocate (=Number of Task Managers)

注意：
	必填的 
	-n   就是 container   =  Number of Task Managers 

不是必填的：
  -jm,--jobManagerMemory <arg>    Memory for JobManager Container with optional unit (default: MB)
 -tm,--taskManagerMemory <arg>   Memory per TaskManager Container with optional unit (default: MB)
    -d,--detached                   If present, runs the job in detached mode     运行你的job 以什么方式

运行起来：

[double_happy@hadoop101 bin]$ ./yarn-session.sh -n 2 -jm 1024 -tm 1024 -d 

2020-01-05 22:37:43,651 INFO  org.apache.flink.configuration.GlobalConfiguration            - Loading configuration property: jobmanager.rpc.address, hadoop101
2020-01-05 22:37:43,656 INFO  org.apache.flink.configuration.GlobalConfiguration            - Loading configuration property: jobmanager.rpc.port, 6123
2020-01-05 22:37:43,656 INFO  org.apache.flink.configuration.GlobalConfiguration            - Loading configuration property: jobmanager.heap.size, 1024m
2020-01-05 22:37:43,657 INFO  org.apache.flink.configuration.GlobalConfiguration            - Loading configuration property: taskmanager.heap.size, 1024m
2020-01-05 22:37:43,657 INFO  org.apache.flink.configuration.GlobalConfiguration            - Loading configuration property: taskmanager.numberOfTaskSlots, 1
2020-01-05 22:37:43,657 INFO  org.apache.flink.configuration.GlobalConfiguration            - Loading configuration property: parallelism.default, 1
2020-01-05 22:37:43,658 INFO  org.apache.flink.configuration.GlobalConfiguration            - Loading configuration property: jobmanager.execution.failover-strategy, region
2020-01-05 22:37:43,658 INFO  org.apache.flink.configuration.GlobalConfiguration            - Loading configuration property: rest.port, 8081
2020-01-05 22:37:44,328 WARN  org.apache.hadoop.util.NativeCodeLoader                       - Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
2020-01-05 22:37:44,505 INFO  org.apache.flink.runtime.security.modules.HadoopModule        - Hadoop user set to double_happy (auth:SIMPLE)
2020-01-05 22:37:44,611 INFO  org.apache.hadoop.yarn.client.RMProxy                         - Connecting to ResourceManager at /0.0.0.0:8032
2020-01-05 22:37:44,761 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                 - The argument n is deprecated in will be ignored.
2020-01-05 22:37:44,930 INFO  org.apache.flink.yarn.AbstractYarnClusterDescriptor           - Cluster specification: ClusterSpecification{masterMemoryMB=1024, taskManagerMemoryMB=1024, numberTaskManagers=2, slotsPerTaskManager=1}
2020-01-05 22:37:45,573 WARN  org.apache.flink.yarn.AbstractYarnClusterDescriptor           - The configuration directory ('/home/double_happy/app/flink-1.9.1/conf') contains both LOG4J and Logback configuration files. Please delete or rename one of them.
2020-01-05 22:37:47,853 INFO  org.apache.flink.yarn.AbstractYarnClusterDescriptor           - Submitting application master application_1578233120766_0002
2020-01-05 22:37:48,255 INFO  org.apache.hadoop.yarn.client.api.impl.YarnClientImpl         - Submitted application application_1578233120766_0002
2020-01-05 22:37:48,255 INFO  org.apache.flink.yarn.AbstractYarnClusterDescriptor           - Waiting for the cluster to be allocated
2020-01-05 22:37:48,257 INFO  org.apache.flink.yarn.AbstractYarnClusterDescriptor           - Deploying cluster, current state ACCEPTED
2020-01-05 22:37:58,915 INFO  org.apache.flink.yarn.AbstractYarnClusterDescriptor           - YARN application has been deployed successfully.
2020-01-05 22:37:58,915 INFO  org.apache.flink.yarn.AbstractYarnClusterDescriptor           - The Flink YARN client has been started in detached mode. In order to stop Flink on YARN, use the following command or a YARN web interface to stop it:
yarn application -kill application_1578233120766_0002
Please also note that the temporary files of the YARN session in the home directory will not be removed.
2020-01-05 22:37:59,734 INFO  org.apache.flink.runtime.rest.RestClient                      - Rest client endpoint started.
Flink JobManager is now running on hadoop101:8081 with leader id 00000000-0000-0000-0000-000000000000.
JobManager Web Interface: http://hadoop101:8081
2020-01-05 22:37:59,757 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                 - The Flink YARN client has been started in detached mode. In order to stop Flink on YARN, use the following command or a YARN web interface to stop it:
yarn application -kill application_1578233120766_0002
[double_happy@hadoop101 bin]$

去yarn 上查看：
在这里插入图片描述

运行一个 例子试试：
	Example

wget -O LICENSE-2.0.txt http://www.apache.org/licenses/LICENSE-2.0.txt
hadoop fs -copyFromLocal LICENSE-2.0.txt hdfs:/// ...
./bin/flink run ./examples/batch/WordCount.jar \
       --input hdfs:///..../LICENSE-2.0.txt --output hdfs:///.../wordcount-result.txt


注意：
	这是啥意思呀：
		就是把数据 上传到hdfs上 然后运行

[double_happy@hadoop101 data]$ hadoop fs -put ./LICENSE-2.0.txt /flink/input/
put: `/flink/input/': No such file or directory

[double_happy@hadoop101 data]$ hadoop fs -mkdir -p  /flink/input/

[double_happy@hadoop101 data]$ hadoop fs -put ./LICENSE-2.0.txt /flink/input/    

[double_happy@hadoop101 data]$ hadoop fs -ls /flink/input/
Found 1 items
-rw-r--r--   1 double_happy supergroup      11358 2020-01-05 22:43 /flink/input/LICENSE-2.0.txt
[double_happy@hadoop101 data]$

运行：

[double_happy@hadoop101 flink]$ ./bin/flink run ./examples/batch/WordCount.jar \
> --input hdfs://hadoop101:8020/flink/input/LICENSE-2.0.txt \
> --output hdfs://hadoop101:8020/flink/output
2020-01-05 22:46:46,517 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                 - Found Yarn properties file under /tmp/.yarn-properties-double_happy.
2020-01-05 22:46:46,517 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                 - Found Yarn properties file under /tmp/.yarn-properties-double_happy.
2020-01-05 22:46:47,119 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                 - YARN properties set default parallelism to 2
2020-01-05 22:46:47,119 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                 - YARN properties set default parallelism to 2
YARN properties set default parallelism to 2
2020-01-05 22:46:47,238 INFO  org.apache.hadoop.yarn.client.RMProxy                         - Connecting to ResourceManager at /0.0.0.0:8032
2020-01-05 22:46:47,449 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                 - No path for the flink jar passed. Using the location of class org.apache.flink.yarn.YarnClusterDescriptor to locate the jar
2020-01-05 22:46:47,449 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                 - No path for the flink jar passed. Using the location of class org.apache.flink.yarn.YarnClusterDescriptor to locate the jar
2020-01-05 22:46:47,650 INFO  org.apache.flink.yarn.AbstractYarnClusterDescriptor           - Found application JobManager host name 'hadoop101' and port '8081' from supplied application id 'application_1578233120766_0002'
Starting execution of program

结果：
[double_happy@hadoop101 flink]$ hadoop fs -ls /flink/output
-rw-r--r--   1 double_happy supergroup       4499 2020-01-05 22:47 /flink/output
[double_happy@hadoop101 flink]$ hadoop fs -text /flink/output
0 3
1 2
2 4
2004 1
3 1
4 1
5 1
50 1
6 1
7 1
8 1
9 2
a 22
above 1
acceptance 1
accepting 3
act 1
acting 1
acts 1
add 2
addendum 1
additional 5
additions 1


注意：
	Flink输出的 是文件 
	Spark是文件夹

注意：
	这个模式 他怎么知道是提交到 哪个session上去呢？

看控制台log：
2020-01-05 22:46:46,517 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli              
   - Found Yarn properties file under /tmp/.yarn-properties-double_happy.

打开： /tmp/.yarn-properties-double_happy


[double_happy@hadoop101 flink]$ cat /tmp/.yarn-properties-double_happy
#Generated YARN properties file
#Sun Jan 05 22:37:59 CST 2020
parallelism=2
dynamicPropertiesString=
applicationID=application_1578233120766_0002     //yarn 的 application的 id
[double_happy@hadoop101 flink]$


注意：
	这块 如果你已经使用了 这个模式跑 之后 还想使用 local跑 
	需要把/tmp/.yarn-properties-double_happy  这个 干掉 也不然 本地模式会 跑到yarn上 报错

Run a single Flink job on YARN
Run a single Flink job on YARN

官方案例：

[double_happy@hadoop101 flink]$ ./bin/flink run -m yarn-cluster ./examples/batch/WordCount.jar
2020-01-05 23:06:06,559 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                 - Found Yarn properties file under /tmp/.yarn-properties-double_happy.
2020-01-05 23:06:06,559 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                 - Found Yarn properties file under /tmp/.yarn-properties-double_happy.
2020-01-05 23:06:07,151 INFO  org.apache.hadoop.yarn.client.RMProxy                         - Connecting to ResourceManager at /0.0.0.0:8032
2020-01-05 23:06:07,324 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                 - No path for the flink jar passed. Using the location of class org.apache.flink.yarn.YarnClusterDescriptor to locate the jar
2020-01-05 23:06:07,324 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                 - No path for the flink jar passed. Using the location of class org.apache.flink.yarn.YarnClusterDescriptor to locate the jar
2020-01-05 23:06:07,520 INFO  org.apache.flink.yarn.AbstractYarnClusterDescriptor           - Cluster specification: ClusterSpecification{masterMemoryMB=1024, taskManagerMemoryMB=1024, numberTaskManagers=1, slotsPerTaskManager=1}
2020-01-05 23:06:08,127 WARN  org.apache.flink.yarn.AbstractYarnClusterDescriptor           - The configuration directory ('/home/double_happy/app/flink-1.9.1/conf') contains both LOG4J and Logback configuration files. Please delete or rename one of them.
2020-01-05 23:06:10,927 INFO  org.apache.flink.yarn.AbstractYarnClusterDescriptor           - Submitting application master application_1578233120766_0003
2020-01-05 23:06:10,965 INFO  org.apache.hadoop.yarn.client.api.impl.YarnClientImpl         - Submitted application application_1578233120766_0003
2020-01-05 23:06:10,966 INFO  org.apache.flink.yarn.AbstractYarnClusterDescriptor           - Waiting for the cluster to be allocated
2020-01-05 23:06:10,971 INFO  org.apache.flink.yarn.AbstractYarnClusterDescriptor           - Deploying cluster, current state ACCEPTED
2020-01-05 23:06:18,402 INFO  org.apache.flink.yarn.AbstractYarnClusterDescriptor           - YARN application has been deployed successfully.
Starting execution of program
Executing WordCount example with default input data set.
Use --input to specify file input.
Printing result to stdout. Use --output to specify output path.
(a,5)
(action,1)
(after,1)
(against,1)
(all,2)
(and,12)
(arms,1)
(arrows,1)
(awry,1)
(ay,1)
(bare,1)
(be,4)
(bear,3)
(bodkin,1)
(bourn,1)
(but,1)
(by,2)
(calamity,1)
(cast,1)
(coil,1)
(come,1)
(conscience,1)
(consummation,1)
(contumely,1)
(country,1)
(cowards,1)
(currents,1)
(d,4)
(death,2)
(delay,1)
(despis,1)
(devoutly,1)
(die,2)
(does,1)
(dread,1)
(dream,1)
(dreams,1)
(end,2)
(enterprises,1)
(er,1)
(fair,1)
(fardels,1)
(flesh,1)
(fly,1)
(for,2)
(fortune,1)
(from,1)
(give,1)
(great,1)
(grunt,1)
(have,2)
(he,1)
(heartache,1)
(heir,1)
(himself,1)
(his,1)
(hue,1)
(ills,1)
(in,3)
(insolence,1)
(is,3)
(know,1)
(law,1)
(life,2)
(long,1)
(lose,1)
(love,1)
(make,2)
(makes,2)
(man,1)
(may,1)
(merit,1)
(might,1)
(mind,1)
(moment,1)
(more,1)
(mortal,1)
(must,1)
(my,1)
(name,1)
(native,1)
(natural,1)
(no,2)
(nobler,1)
(not,2)
(now,1)
(nymph,1)
(o,1)
(of,15)
(off,1)
(office,1)
(ophelia,1)
(opposing,1)
(oppressor,1)
(or,2)
(orisons,1)
(others,1)
(outrageous,1)
(pale,1)
(pangs,1)
(patient,1)
(pause,1)
(perchance,1)
(pith,1)
(proud,1)
(puzzles,1)
(question,1)
(quietus,1)
(rather,1)
(regard,1)
(remember,1)
(resolution,1)
(respect,1)
(returns,1)
(rub,1)
(s,5)
(say,1)
(scorns,1)
(sea,1)
(shocks,1)
(shuffled,1)
(sicklied,1)
(sins,1)
(sleep,5)
(slings,1)
(so,1)
(soft,1)
(something,1)
(spurns,1)
(suffer,1)
(sweat,1)
(take,1)
(takes,1)
(than,1)
(that,7)
(the,22)
(their,1)
(them,1)
(there,2)
(these,1)
(this,2)
(those,1)
(thought,1)
(thousand,1)
(thus,2)
(thy,1)
(time,1)
(tis,2)
(to,15)
(traveller,1)
(troubles,1)
(turn,1)
(under,1)
(undiscover,1)
(unworthy,1)
(us,3)
(we,4)
(weary,1)
(what,1)
(when,2)
(whether,1)
(whips,1)
(who,2)
(whose,1)
(will,1)
(wish,1)
(with,3)
(would,2)
(wrong,1)
(you,1)
Program execution finished
Job with JobID 0af373b2d338811166328e7e1fca1fe1 has finished.
Job Runtime: 10995 ms
Accumulator Results: 
- fed80ecbe6107c17759a1be247abbdd4 (java.util.ArrayList) [170 elements]

[double_happy@hadoop101 flink]$

运行一下之前的local的程序：

运行一下 之前的local的 程序：
[double_happy@hadoop101 flink]$ ./bin/flink run \
> --class com.sx.flink04.SocketWindowWordCount \
> -m yarn-cluster \
> /home/double_happy/lib/Flink-1.0.jar \
> --port 9998 \
> --host hadoop101
2020-01-05 23:08:37,619 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                 - Found Yarn properties file under /tmp/.yarn-properties-double_happy.
2020-01-05 23:08:37,619 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                 - Found Yarn properties file under /tmp/.yarn-properties-double_happy.
2020-01-05 23:08:38,076 INFO  org.apache.hadoop.yarn.client.RMProxy                         - Connecting to ResourceManager at /0.0.0.0:8032
2020-01-05 23:08:38,294 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                 - No path for the flink jar passed. Using the location of class org.apache.flink.yarn.YarnClusterDescriptor to locate the jar
2020-01-05 23:08:38,294 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                 - No path for the flink jar passed. Using the location of class org.apache.flink.yarn.YarnClusterDescriptor to locate the jar
2020-01-05 23:08:38,467 INFO  org.apache.flink.yarn.AbstractYarnClusterDescriptor           - Cluster specification: ClusterSpecification{masterMemoryMB=1024, taskManagerMemoryMB=1024, numberTaskManagers=1, slotsPerTaskManager=1}
2020-01-05 23:08:39,079 WARN  org.apache.flink.yarn.AbstractYarnClusterDescriptor           - The configuration directory ('/home/double_happy/app/flink-1.9.1/conf') contains both LOG4J and Logback configuration files. Please delete or rename one of them.
2020-01-05 23:08:40,769 INFO  org.apache.flink.yarn.AbstractYarnClusterDescriptor           - Submitting application master application_1578233120766_0004
2020-01-05 23:08:41,006 INFO  org.apache.hadoop.yarn.client.api.impl.YarnClientImpl         - Submitted application application_1578233120766_0004
2020-01-05 23:08:41,007 INFO  org.apache.flink.yarn.AbstractYarnClusterDescriptor           - Waiting for the cluster to be allocated
2020-01-05 23:08:41,010 INFO  org.apache.flink.yarn.AbstractYarnClusterDescriptor           - Deploying cluster, current state ACCEPTED
2020-01-05 23:08:46,914 INFO  org.apache.flink.yarn.AbstractYarnClusterDescriptor           - YARN application has been deployed successfully.
Starting execution of program

在这里插入图片描述

输入数据：
[double_happy@hadoop101 data]$ nc -lk 9998
啊，a,a,a,v,b,c
c,c,d,d,,b,b


查看结果：

在这里插入图片描述

这是yarn的两种方式：

我使用的是第二种方式 ：
	如果需要一个MySQL jar包该怎么传？
		我接触flink 时间很短  我身边同事也不知道 他们经常搞Flink的 竟然不知道？ 就知道打胖包？ 绝对是不可以的

Flink03--double_happy

2020-01-05

记录

写东西 还是要有通用的思想

通用的！！！ 与底层的执行引擎是没关系的  Beam
    Spark
        1.6
        2.x
    Flink

    你的API上一定不能出现底层执行引擎的API

    适配 ==> Spark/Flink

注意

上一篇文章 ：
MySQLSource：
class MySQLSource extends RichSourceFunction[Student]{

  var connection:Connection = _
  var pstmt:PreparedStatement = _

  // 在open方法中建立连接
  override def open(parameters: Configuration): Unit = {
    super.open(parameters)

    connection = MySQLUtils.getConnection("hadoop101","3306","test","root", "wsx123$%^")
    pstmt = connection.prepareStatement("select * from student")
  }

  // 释放
  override def close(): Unit = {
    super.close()
    MySQLUtils.closeConnection(connection, pstmt)
  }

  override def cancel(): Unit = {

  }

  override def run(ctx: SourceFunction.SourceContext[Student]): Unit = {

    println("~~~~run~~~~~~")
    val rs = pstmt.executeQuery()
    while(rs.next()){
      val student = Student(rs.getInt("id"), rs.getString("name"), rs.getString("password"),rs.getInt("age"))
      ctx.collect(student)
    }
  }
}

注意：
class MySQLSource extends RichSourceFunction[Student]

RichSourceFunction：
这个东西 并不是 上一篇文章的三个Function 中的一个 

那么 RichSourceFunction 的并行度是多少？


源码：
public abstract class RichSourceFunction<OUT> extends AbstractRichFunction implements SourceFunction<OUT> {

	private static final long serialVersionUID = 1L;
}

我当时测试的时候 设置并行度为3的时候 报错的 

源码里是继承SourceFunction的 
所以并行度是1

那么就有一个问题？
 有两个流：
 
	stream1  4并行度  emp表  log里      domain
	stream2  1并行度  dept表  MySQL里    user_id
这两个流进行 
stream1.connect(stream2).map(x=>{})

即：
	logStream emp: deptno empno ename
	mysqlStream dept: deptno dname
	
	outputStream empno,ename,deptno,dname
	
	数据清洗/ETL
	outputStream = stream1.connect(stream2).map(x=>{
	    ....
	})

那么connect会成功么？？
	4并行度的流 里面有写task 的deptno 关联 
	1并行度的流 拿取dname 是拿不到的 
那么这个问题该怎么解决呢？之后再说

先解决并行度为1的问题 下面
 变为：RichParallelSourceFunction

class ScalikeJDBCMySQLSource extends RichParallelSourceFunction[Student]{

  override def cancel(): Unit = {

  }

  override def run(ctx: SourceFunction.SourceContext[Student]): Unit = {
    println("~~~run~~~~")
    DBs.setupAll()  // parse configuration file

    DB.readOnly{ implicit session => {
      SQL("select * from student").map(rs => {
        val student = Student(rs.int("id"),rs.string("name"),rs.string("password"),rs.int("age"))
        ctx.collect(student)
      }).list().apply()
    }

    }

  }
}

object SourceApp {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment

     env.addSource(new ScalikeJDBCMySQLSource).setParallelism(2).print()

    env.execute(this.getClass.getSimpleName)
  }
}

结果是：
~~~run~~~~
~~~run~~~~
6> Student(1,kairis,wsx111,17)
1> Student(4,happy,11,48)
8> Student(3,double,44,12)
7> Student(2,dbh,11,90)
7> Student(3,double,44,12)
5> Student(1,kairis,wsx111,17)
6> Student(2,dbh,11,90)
8> Student(4,happy,11,48)

2并行度 也就是结果会拿取两份

第二个问题：
接着最开始流connect的问题 

1.log流 的并行度是4   （100w条）
那么 MySQL里的并行度是否有必要也是4呢？ （MySQl 1w条）


结果：
 是不是 没有这个必要呀 
 1个并行度能处理过来的 就可以了 
 但是1个并行度 也log流4个并行度 connect的时候还会出问题
那么该怎么办呢？之后说

Data Sinks

输出对比：
Spark读写：外部数据源
    spark.read.format("").option("",").load
    spark.write.format("")....save

Flink读写
    addSource(new XXXSourceFunction)
    addSink(new XXXSinkFunction)

   都是  可插拔的  Spark 自定义外部数据源 
还是Spark好用呀

案例

Sink kafka

Flink的角色是Producer：

object SinkApp {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment

        val stream = env.readTextFile("C:\\IdeaProjects\\flink\\data\\access.log").map(x => {
          val splits = x.split(",")
          Access(splits(0).toLong, splits(1), splits(2).toLong).toString
        })

        val producer = new FlinkKafkaProducer[String](
          "hadoop101:9092,hadoop101:9093,hadoop101:9094",         // broker list
          "double_happy_offset",               // target topic
          new SimpleStringSchema)   // serialization schema

        stream.addSink(producer)  // 2Kafka
        stream.print() // 2Local


    env.execute(this.getClass.getSimpleName)
  }
}
结果是：
[double_happy@hadoop101 kafka]$ bin/kafka-console-consumer.sh \
> --bootstrap-server hadoop101:9092,hadoop101:9093,hadoop101:9094 \
> --topic double_happy_offset 
Access(201912120010,ruozedata.com,2000)
Access(201912120010,ruozedata.com,4000)
Access(201912120010,dongqiudi.com,1000)
Access(201912120010,dongqiudi.com,6000)
Access(201912120010,zhibo8.com,5000)

Kafak 2 Kafka

object SinkApp {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    val properties = new Properties()
    properties.setProperty("bootstrap.servers",  "hadoop101:9092,hadoop101:9093,hadoop101:9094")
    properties.setProperty("group.id", "sxwang")
    
    val consumer = new FlinkKafkaConsumer[String]("double_happy_offset", new SimpleStringSchema(), properties)
    
    val stream = env.addSource(consumer)

    // TODO... Kafka2Kafka  double_happy_offset ==> double_happy_offset_test
    
    val producer = new FlinkKafkaProducer[String](
      "hadoop101:9092,hadoop101:9093,hadoop101:9094", // broker list
      "double_happy_offset_test", // target topic
      new SimpleStringSchema) // serialization schema

    stream.addSink(producer) // 2Kafka
    stream.print() // 2Local
    
    env.execute(this.getClass.getSimpleName)
  }

}

结果是：
[double_happy@hadoop101 kafka]$ bin/kafka-console-consumer.sh \
> --bootstrap-server hadoop101:9092,hadoop101:9093,hadoop101:9094 \
> --topic double_happy_offset_test 
Access(201912120010,ruozedata.com,2000)
Access(201912120010,dongqiudi.com,1000)
Access(201912120010,dongqiudi.com,6000)
e
c
Access(201912120010,ruozedata.com,4000)
b
e
Access(201912120010,zhibo8.com,5000)
c
f

Sink MySQL

/**
 * A {@link org.apache.flink.api.common.functions.RichFunction} version of {@link SinkFunction}.
 */
@Public
public abstract class RichSinkFunction<IN> extends AbstractRichFunction implements SinkFunction<IN> {

	private static final long serialVersionUID = 1L;
}



/**
 * Interface for implementing user defined sink functionality.
 *
 * @param <IN> Input type parameter.
 */
@Public
public interface SinkFunction<IN> extends Function, Serializable {


所以
IN： 表示 Input type parameter. 

表示sink到的 MySQL表里的 字段类型

class DoubleHappyMySQLSink  extends RichSinkFunction[(String,Int)]{

  var connection:Connection = _
  var insertPstmt:PreparedStatement = _
  var updatePstmt:PreparedStatement = _

  // 打开connection等
  override def open(parameters: Configuration): Unit = {
    super.open(parameters)

    /**
      * 这块与Spark不一样的 Spark是一个批次往下写 
      * 
      * 而Flink是 来一条就往下写  
      * 
      * 所以 :
      *   这块创建两个 PreparedStatement 也可以创建一个 需要创建表的时候指定 key 即可 
      *   
      *   创建 两个简单些  ： 有就更新 没有就插入 
      */
    connection = MySQLUtils.getConnection()
    insertPstmt = connection.prepareStatement("insert into domain(domain,traffic) values (?,?)")
    updatePstmt = connection.prepareStatement("update domain set traffic=? where domain=?")
  }

  /**
    * 写数据
    */
  override def invoke(value: (String, Int), context: SinkFunction.Context[_]): Unit = {
    // TODO   insert update

    /**
      * 这块并没有写错哦  要看 你的sql语句的 ？ 走的哦 
      * 
      * 先更新 有就更新 无就插入
      */
    updatePstmt.setInt(1, value._2)
    updatePstmt.setString(2, value._1)
    updatePstmt.execute()

    if(updatePstmt.getUpdateCount == 0) {
      insertPstmt.setString(1, value._1)
      insertPstmt.setInt(2, value._2)
      insertPstmt.execute()
    }
  }


  // 释放资源
  override def close(): Unit = {
    super.close()
    if(insertPstmt != null) insertPstmt.close()
    if(updatePstmt != null) updatePstmt.close()
    if(connection != null) connection.close()
  }

}


注意：
这代码不严谨的

object SinkApp {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    val stream = env.readTextFile("C:\\IdeaProjects\\flink\\data\\access.log").map(x => {
      val splits = x.split(",")
      (splits(1), splits(2).toInt)
    }).keyBy(0).sum(1)

    stream.addSink(new DoubleHappyMySQLSink)


    env.execute(this.getClass.getSimpleName)
  }

}

结果是： 我执行了两遍Flink程序  发现 结果幂等性 ok
mysql> select * from domain;
+---------------+---------+
| domain        | traffic |
+---------------+---------+
| dongqiudi.com |    7000 |
| ruozedata.com |    6000 |
| zhibo8.com    |    5000 |
+---------------+---------+
3 rows in set (0.00 sec)

mysql> select * from domain;
+---------------+---------+
| domain        | traffic |
+---------------+---------+
| dongqiudi.com |    7000 |
| ruozedata.com |    6000 |
| zhibo8.com    |    5000 |
+---------------+---------+
3 rows in set (0.00 sec)

mysql>

Debugging ***

调试的手段***

Iterator Data Sink
Iterator Data Sink

注意：
	你调试的时候 ：
		1.不需要接kafka   使用 自定义Source 就可以 eg：RichParallelSourceFunction
		2.Sink的时候 也不需要Sink 到 存储的地方去
			eg：kudu 、Hbase等
			这里使用 Iterator Data Sink
这个真的很重要：
	我封闭开发的时候 字段154个 Sink到kudu里面 
如果使用 提供的 connecer 154个字段 你什么时候能写完 
	一定是要写成 自动解析的 最好 

注意：
	刚好 我的业务场景 需要这个 Iterator Data Sink

官网：
import org.apache.flink.streaming.experimental.DataStreamUtils
import scala.collection.JavaConverters.asScalaIteratorConverter

val myResult: DataStream[(String, Int)] = ...
val myOutput: Iterator[(String, Int)] = DataStreamUtils.collect(myResult.javaStream).asScala

案例：

object SourceApp {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    //测试使用

    val data: DataStream[Domain.Access] = env.addSource(new AccessSource03).setParallelism(1)

    import scala.collection.JavaConverters.asScalaIteratorConverter
    val output: Iterator[Domain.Access] = DataStreamUtils.collect(data.javaStream).asScala
    
    output.take(1).foreach(println(_))

    env.execute(this.getClass.getSimpleName)
  }
}

结果：
Access(1577975577015,zhibo8.cc,78)

Sink redis

connector
在这里插入图片描述

注意：
pom 选择的 版本;

<dependency>
  <groupId>org.apache.bahir</groupId>
  <artifactId>flink-connector-redis_2.11</artifactId>
  <version>1.1-SNAPSHOT</version>
</dependency>

这个版本可能下载不到 ：

在这里插入图片描述

所以下载：
<!-- https://mvnrepository.com/artifact/org.apache.bahir/flink-connector-redis -->
<dependency>
    <groupId>org.apache.bahir</groupId>
    <artifactId>flink-connector-redis_2.11</artifactId>
    <version>1.0</version>
</dependency>

Flink Redis Connector

class DoubleHappyRedisSink  extends RedisMapper[(String,Int)]{
  override def getCommandDescription: RedisCommandDescription = {
    new RedisCommandDescription(RedisCommand.HSET, "double_happy_traffic")
  }

  override def getValueFromData(data: (String, Int)): String = {
    data._2 + ""
  }

  override def getKeyFromData(data: (String, Int)): String = {
    data._1
  }
}

object SinkApp {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    val stream = env.readTextFile("C:\\IdeaProjects\\flink\\data\\access.log").map(x => {
      val splits = x.split(",")
      (splits(1), splits(2).toInt)
    }).keyBy(0).sum(1)


    val conf = new FlinkJedisPoolConfig.Builder().setHost("hadoop101").build()

    stream.addSink(new RedisSink[(String, Int)](conf, new DoubleHappyRedisSink))

    env.execute(this.getClass.getSimpleName)
  }

}
结果是：
hadoop101:6379> HGETALL double_happy_traffic
1) "ruozedata.com"
2) "6000"
3) "dongqiudi.com"
4) "7000"
5) "zhibo8.com"
6) "5000"
hadoop101:6379> 

与前面的MySQL 结果是一样的

***解决前面的问题

1
2

Flink02--double_happy

2020-01-05

概述

Function
    map ==> MapFunction
    filter ==> FilterFunction
    xxx ==>  XxxFunction
    RichXxxFunction  *****

SourceFunction     non-parallel 1
ParallelSourceFunction
RichParallelSourceFunction  *****

//测试使用

object SourceApp {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    //测试使用
    env.fromCollection(List(
      Access(201912120010L, "ruozedata.com", 2000),
      Access(201912120011L, "ruozedata.com", 3000)
    )
    ).print()

    env.execute(this.getClass.getSimpleName)
  }
}

结果：
6> Access(201912120011,ruozedata.com,3000)
5> Access(201912120010,ruozedata.com,2000)

/**
   * Creates a DataStream that contains the given elements. The elements must all be of the
   * same type.
   *
   * Note that this operation will result in a non-parallel data source, i.e. a data source with
   * a parallelism of one.
   */
  def fromElements[T: TypeInformation](data: T*): DataStream[T] = {
    fromCollection(data)
  }

注意：
	前面的泛型 如果加上 会类型限定的

object SourceApp {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    //测试使用
    env.fromElements(1, 2L, 3D, 4F, "5").print()

    env.execute(this.getClass.getSimpleName)
  }
}

结果：
1> 4.0
8> 3.0
2> 5
7> 2
6> 1

注意：
	如果加上泛型  会类型限定的

在这里插入图片描述
Custom:

env.addSource()

1.SourceFunction 
    cancel()
    run()
注意：
	1.自定义数据源
	      继承SourceFunction    要传一个泛型
	2.实现 下面这两个方法
		cancel()
			用于关闭某个东西
        run()  里面有ctx 上下文
          用于产生数据的 
	         因为这是SourceFunction 是数据源
	         可以利用ctx 产生数据的

注意：
	SourceFunction 是没有并行度可言的

这个很重要用于测试产生数据

package com.sx.flink02

import com.sx.bean.Domain.Access
import org.apache.flink.streaming.api.functions.source.SourceFunction

import scala.util.Random

class AccessSource extends SourceFunction[Access]{

  var running = true

  override def cancel(): Unit = {
    running = false
  }

  override def run(ctx: SourceFunction.SourceContext[Access]): Unit = {
    val random = new Random()
    val domains = Array("ruozedata.com","zhibo8.cc","dongqiudi.com")

    while(running) {
      val timestamp = System.currentTimeMillis()
      1.to(10).map(x => {
        ctx.collect(Access(timestamp, domains(random.nextInt(domains.length)), random.nextInt(1000)+x))
      })

      // 休息下 每个5s
      Thread.sleep(5000)
    }
  }
}

object SourceApp {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    //测试使用
    env.addSource(new AccessSource).print()
    
    env.execute(this.getClass.getSimpleName)
  }
}

结果是：
7> Access(1577502637884,zhibo8.cc,991)
2> Access(1577502637884,ruozedata.com,252)
5> Access(1577502637884,dongqiudi.com,428)
8> Access(1577502637884,dongqiudi.com,565)
4> Access(1577502637884,ruozedata.com,140)
6> Access(1577502637884,ruozedata.com,718)
1> Access(1577502637884,dongqiudi.com,66)
3> Access(1577502637884,dongqiudi.com,475)
4> Access(1577502637884,zhibo8.cc,694)
3> Access(1577502637884,zhibo8.cc,972)

这就是每个5s发送10条数据

测试:
SourceFunction 是没有并行度可言的 
object SourceApp {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    //测试使用

    env.addSource(new AccessSource).setParallelism(3).print()

    env.execute(this.getClass.getSimpleName)
  }
}
结果：
Exception in thread "main" java.lang.IllegalArgumentException: Source: 1 is not a parallel source
	at org.apache.flink.streaming.api.datastream.DataStreamSource.setParallelism(DataStreamSource.java:55)
	at org.apache.flink.streaming.api.datastream.DataStreamSource.setParallelism(DataStreamSource.java:31)
	at org.apache.flink.streaming.api.scala.DataStream.setParallelism(DataStream.scala:130)
	at com.sx.flink02.SourceApp$.main(SourceApp.scala:15)
	at com.sx.flink02.SourceApp.main(SourceApp.scala)

注意：
Source: 1 is not a parallel source

查看源码：
debug查看

在这里插入图片描述

所以在大数据场景下 没有并行度 这个就不能用

2.ParallelSourceFunction  带并行度的 

class AccessSource02 extends ParallelSourceFunction[Access]{

  var running = true

  override def cancel(): Unit = {
    running = false
  }

  override def run(ctx: SourceFunction.SourceContext[Access]): Unit = {
    val random = new Random()
    val domains = Array("ruozedata.com","zhibo8.cc","dongqiudi.com")

    while(running) {
      val timestamp = System.currentTimeMillis()
      1.to(10).map(x => {
        ctx.collect(Access(timestamp, domains(random.nextInt(domains.length)), random.nextInt(1000)+x))
      })

      // 休息下
      Thread.sleep(5000)
    }
  }
}

注意：
代码是一样的 就换个继承

object SourceApp {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    //测试使用
    env.addSource(new AccessSource02).setParallelism(3).print()

    env.execute(this.getClass.getSimpleName)
  }
}

结果是：
6> Access(1577503201868,dongqiudi.com,704)
8> Access(1577503201868,zhibo8.cc,562)
2> Access(1577503201868,dongqiudi.com,526)
7> Access(1577503201868,ruozedata.com,952)
3> Access(1577503201868,ruozedata.com,525)
2> Access(1577503201868,dongqiudi.com,728)
7> Access(1577503201868,dongqiudi.com,185)
5> Access(1577503201868,ruozedata.com,737)
2> Access(1577503201868,zhibo8.cc,47)
1> Access(1577503201868,dongqiudi.com,77)
4> Access(1577503201868,ruozedata.com,275)
5> Access(1577503201868,ruozedata.com,47)
7> Access(1577503201868,dongqiudi.com,289)
3> Access(1577503201868,dongqiudi.com,934)
8> Access(1577503201868,ruozedata.com,69)
6> Access(1577503201868,dongqiudi.com,123)
4> Access(1577503201868,zhibo8.cc,719)
1> Access(1577503201868,zhibo8.cc,254)
6> Access(1577503201868,zhibo8.cc,1002)
3> Access(1577503201868,zhibo8.cc,358)
7> Access(1577503201868,dongqiudi.com,700)
5> Access(1577503201868,dongqiudi.com,692)
6> Access(1577503201868,ruozedata.com,408)
5> Access(1577503201868,ruozedata.com,986)
1> Access(1577503201868,ruozedata.com,284)
4> Access(1577503201868,zhibo8.cc,900)
5> Access(1577503201868,ruozedata.com,343)
8> Access(1577503201868,ruozedata.com,84)
4> Access(1577503201868,zhibo8.cc,967)
8> Access(1577503201868,ruozedata.com,14)

注意：
这就是每个5s发送10条数据 但是并行度是3的情况下 
就是 每个5s 发送 10*3 = 30 条数据

3.RichParallelSourceFunction ****
注意：
	和上面的核心代码一样
	但是 这个Rich里面

/**
 * Base class for implementing a parallel data source. Upon execution, the runtime will
 * execute as many parallel instances of this function as configured parallelism
 * of the source.
 *
 * <p>The data source has access to context information (such as the number of parallel
 * instances of the source, and which parallel instance the current instance is)
 * via {@link #getRuntimeContext()}. It also provides additional life-cycle methods
 * ({@link #open(org.apache.flink.configuration.Configuration)} and {@link #close()}.</p>
 *
 * @param <OUT> The type of the records produced by this source.
 */
@Public
public abstract class RichParallelSourceFunction<OUT> extends AbstractRichFunction
		implements ParallelSourceFunction<OUT> {

	private static final long serialVersionUID = 1L;
}


1. extends AbstractRichFunction 
而AbstractRichFunction 
是有生命周期的方法
    open()
    close()

所以：
	如果做一些
	1.文件系统 
	2.初始化
	3.io
	4.mysql 等等
就得在open里和close里面获取连接 和关闭连接

因为：
	open()  close() 一个task会执行一次 （前面文章有例子）task就是并行度
	所以要使用这个Function 
	生命周期是可以控制的

class AccessSource03 extends RichParallelSourceFunction[Access]{

  var running = true

  override def cancel(): Unit = {
    running = false
  }

  override def open(parameters: Configuration): Unit = {
    super.open(parameters)
  }

  override def close(): Unit = {
    super.close()
  }
  
  override def run(ctx: SourceFunction.SourceContext[Access]): Unit = {
    val random = new Random()
    val domains = Array("ruozedata.com","zhibo8.cc","dongqiudi.com")

    while(running) {
      val timestamp = System.currentTimeMillis()
      1.to(10).map(x => {
        ctx.collect(Access(timestamp, domains(random.nextInt(domains.length)), random.nextInt(1000)+x))
      })

      // 休息下
      Thread.sleep(5000)
    }
  }
}

上面就是 Source的最基本的使用

那么现在有一个需求：
  连接MySQL 
  查看官网 没有MySQL的Connector

那么只能去github上找 或者 自己写一个Source即可

需求就是：
  把MySQL数据读取进来 

操作MySQL pom里要加入MySQL依赖的

MySQL里面的数据：
mysql> select * from student;
+----+--------+----------+-----+
| id | name   | password | age |
+----+--------+----------+-----+
|  1 | kairis | wsx111   |  17 |
|  2 | dbh    | 11       |  90 |
|  3 | double | 44       |  12 |
|  4 | happy  | 11       |  48 |
+----+--------+----------+-----+
4 rows in set (0.00 sec)

mysql>

class MySQLSource extends RichSourceFunction[Student]{
  var connection:Connection = _
  var pstmt:PreparedStatement = _
  // 在open方法中建立连接
  override def open(parameters: Configuration): Unit = {
    super.open(parameters)
    connection = MySQLUtils.getConnection()
    pstmt = connection.prepareStatement("select * from student")
  }

  // 释放
  override def close(): Unit = {
    super.close()
    MySQLUtils.closeConnection(connection, pstmt)
  }
  override def cancel(): Unit = {

  }
  override def run(ctx: SourceFunction.SourceContext[Student]): Unit = {
    val rs = pstmt.executeQuery()
    while(rs.next()){
      val student = Student(rs.getInt("id"), rs.getString("name"), rs.getString("password"),rs.getInt("age"))
      ctx.collect(student)
    }
  }
}

注意：
	连接MySQL 生产上是有并行度的 
	你不能在run里面建立连接   为什么呢？什么会触发run呢？主要的逻辑会触发run 
	这就类似Spark里面的 foreach 和foreachPartiion
得借助于 open方法：
	 去拿到连接   这个是每个task会执行一次

object SourceApp {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.addSource(new MySQLSource).print()
    env.execute(this.getClass.getSimpleName)
  }
}

结果：
~~~~run~~~~~~
5> Student(4,happy,11,48)
4> Student(3,double,44,12)
3> Student(2,dbh,11,90)
2> Student(1,kairis,wsx111,17)

env.addSource(new MySQLSource).setParallelism(3).print()
会报错的：
Exception in thread "main" java.lang.IllegalArgumentException: Source: 1 is not a parallel source
为什么呢？
但是这种方式 太low了 


使用scalikjdbc： 这种方式更优雅些
在配置文件里可以配置连接池

class ScalikeJDBCMySQLSource extends RichSourceFunction[Student]{

  override def cancel(): Unit = {

  }

  override def run(ctx: SourceFunction.SourceContext[Student]): Unit = {
    println("~~~run~~~~")
    DBs.setupAll()  // parse configuration file

    DB.readOnly{ implicit session => {
      SQL("select * from student").map(rs => {
        val student = Student(rs.int("id"),rs.string("name"),rs.string("password"),rs.int("age"))
        ctx.collect(student)
      }).list().apply()
    }

    }
  }
}

结果：
~~~run~~~~
7> Student(4,happy,11,48)
5> Student(2,dbh,11,90)
4> Student(1,kairis,wsx111,17)
6> Student(3,double,44,12)

读取kafka的数据

注意：
不同的kafka版本 Flink 使用的api不同

在这里插入图片描述

但是在 0.11之后就统一了

在这里插入图片描述

我的kafka是 2.2.1系列的

所以我使用的 pom是：
我的flink 是 1.9.0的
<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-connector-kafka_2.11</artifactId>
  <version>1.9.0</version>
</dependency>

Flink’s Kafka consumer is
 called FlinkKafkaConsumer08 (or 09 for
  Kafka 0.9.0.x versions, etc. 
 or 
 just FlinkKafkaConsumer for Kafka >= 1.0.0 versions).
  It provides access to one or more Kafka topics.



Source：
	Flink作为Kafka的消费者

/**
	 * Creates a new Kafka streaming source consumer.
	 *
	 * @param topic             The name of the topic that should be consumed.
	 * @param valueDeserializer The de-/serializer used to convert between Kafka's byte messages and Flink's objects.
	 * @param props
	 */
	public FlinkKafkaConsumer(String topic, DeserializationSchema<T> valueDeserializer, Properties props) {
		this(Collections.singletonList(topic), valueDeserializer, props);
	}

1.topic
2.DeserializationSchema   接收数据 所以是反序列化器 消费数据 所以是反序列化的
3.Properties

[double_happy@hadoop101 kafka]$ jps
Kafka
Kafka
Jps
QuorumPeerMain
Kafka

先开启：
object SourceApp {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    val properties = new Properties()
    properties.setProperty("bootstrap.servers", "hadoop101:9092,hadoop101:9093,hadoop101:9094")
    properties.setProperty("group.id", "sxwang")

    val consumer = new FlinkKafkaConsumer[String]("double_happy_offset", new SimpleStringSchema(), properties)

    env.addSource(consumer).print()

    env.execute(this.getClass.getSimpleName)
  }
}
结果：
1> b
2> d
3> d
1> c
2> f
3> b

往kafka发送数据：
object DataGenerator {

  private val logger: Logger = LoggerFactory.getLogger(DataGenerator.getClass)


  def main(args: Array[String]): Unit = {

    val props = new Properties()
    props.put("bootstrap.servers", "hadoop101:9092,hadoop101:9093,hadoop101:9094")
    props.put("acks", "all")
    props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")
    props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer")
    val producer = new KafkaProducer[String, String](props)

    for (i <- 0 to 5) {
      Thread.sleep(100)

      //拿一个abcdef
      val word: String = String.valueOf((new Random().nextInt(6) + 'a').toChar)
      val part = i % 3 //发到哪个分区 因为是三个分区

      logger.error("word : {}", word)

      val record = producer.send(new ProducerRecord[String, String]("double_happy_offset", part, "",word))

    }

    producer.close()
    println("double_happy 数据产生完毕..........")
  }
}

Flink里面的kafka的offset 非常 好管理的 
Spark里面的 是批次 每个批次的偏移量的管理 
Flink是一条数据 进来的 一条数据进来的 
控制偏移量就是api的使用 很简单

val myConsumer = new FlinkKafkaConsumer08[String](...)
myConsumer.setStartFromEarliest()      // start from the earliest record possible
myConsumer.setStartFromLatest()        // start from the latest record
myConsumer.setStartFromTimestamp(...)  // start from specified epoch timestamp (milliseconds)
myConsumer.setStartFromGroupOffsets()  // the default behaviour

val stream = env.addSource(myConsumer)

这是官网上的

那么先讲使用 里面的状态和checkpoint 容错等 后续写入文章

object SourceApp {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    val properties = new Properties()
    properties.setProperty("bootstrap.servers", "hadoop101:9092,hadoop101:9093,hadoop101:9094")
    properties.setProperty("group.id", "sxwang")

    val consumer = new FlinkKafkaConsumer[String]("double_happy_offset", new SimpleStringSchema(), properties)

    consumer.setStartFromEarliest()

    env.addSource(consumer).print()

    env.execute(this.getClass.getSimpleName)
  }
}

结果：
1> a
2> f
3> b
2> f
1> b
2> d
3> e
2> f
1> c
3> d
3> b

看 这就把我们前面写入kafka的数据读出来了

所以Flink里的偏移量很简单 底层管理的很好

问题：
 我能不能指定 一个分区 分区里面的 offset呢？

在这里插入图片描述
DataStream Transformations

和Spark大部分都差不多 
下面说一下 Spark里没有的

在这里插入图片描述

object TranformationApp {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment

        val stream = env.readTextFile("C:\\IdeaProjects\\flink\\data\\access.log").map(x => {
          val splits = x.split(",")
          Access(splits(0).toLong, splits(1), splits(2).toLong)
        })

    stream.keyBy(_.domain).sum("traffic").print("sum")

    env.execute(this.getClass.getSimpleName)
  }
}
结果是：
sum:5> Access(201912120010,ruozedata.com,2000)
sum:6> Access(201912120010,dongqiudi.com,1000)
sum:6> Access(201912120010,zhibo8.com,5000)
sum:5> Access(201912120010,ruozedata.com,6000)
sum:6> Access(201912120010,dongqiudi.com,7000)

为什么结果是这样子的：
   因为这是进来一条统计一次 
注意和Spark的区别： Spark里是一个批次的

object TranformationApp {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment


        val stream = env.readTextFile("C:\\IdeaProjects\\flink\\data\\access.log").map(x => {
          val splits = x.split(",")
          Access(splits(0).toLong, splits(1), splits(2).toLong)
        })

        stream.keyBy(_.domain).reduce((x,y) => {
          Access(x.time, x.domain, (x.traffic+y.traffic+100))
        }).print()

    env.execute(this.getClass.getSimpleName)
  }

}

结果：
5> Access(201912120010,ruozedata.com,4000)
6> Access(201912120010,dongqiudi.com,1000)
6> Access(201912120010,dongqiudi.com,7100)
5> Access(201912120010,ruozedata.com,6100)
6> Access(201912120010,zhibo8.com,5000)

注意：
1.reduce 比sum灵活   keyby之后 是把相同的domain 放到一起了
2.为什么zhibo8.com 结果没有＋100

进来一条统计一次  为什么结果没有+100呢？
因为 就1个呀   reduce 把相邻的两个做操作 1个操作啥

分流：

在这里插入图片描述

这个功能 可能会用的到：
这里使用split 后续 是使用  侧输出  因为split过时了  不过没有关系

object TranformationApp {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    val stream = env.readTextFile("C:\\IdeaProjects\\flink\\data\\access.log").map(x => {
      val splits = x.split(",")
      Access(splits(0).toLong, splits(1), splits(2).toLong)
    })

    // 5000 6000 7000
    val splitStream = stream.keyBy("domain").sum("traffic").split(x => {
      if (x.traffic > 6000) {
        Seq("大客户")
      } else {
        Seq("一般客户")
      }
    })

    splitStream.select("大客户").print("大客户")
    splitStream.select("一般客户").print("一般客户")
  
    env.execute(this.getClass.getSimpleName)
  }
}

结果是：
一般客户:5> Access(201912120010,ruozedata.com,2000)
一般客户:6> Access(201912120010,dongqiudi.com,1000)
一般客户:6> Access(201912120010,zhibo8.com,5000)
一般客户:5> Access(201912120010,ruozedata.com,6000)
大客户:6> Access(201912120010,dongqiudi.com,7000)

splitStream.select("大客户", "一般客户").print("ALL")

结果是：
一般客户:5> Access(201912120010,ruozedata.com,2000)
一般客户:6> Access(201912120010,dongqiudi.com,1000)
ALL:5> Access(201912120010,ruozedata.com,2000)
ALL:6> Access(201912120010,dongqiudi.com,1000)
一般客户:6> Access(201912120010,zhibo8.com,5000)
ALL:6> Access(201912120010,zhibo8.com,5000)
一般客户:5> Access(201912120010,ruozedata.com,6000)
大客户:6> Access(201912120010,dongqiudi.com,7000)
ALL:5> Access(201912120010,ruozedata.com,6000)
ALL:6> Access(201912120010,dongqiudi.com,7000)

合流

两个流是可以合的 ：
union
connect

object TranformationApp {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment


    val stream = env.readTextFile("C:\\IdeaProjects\\flink\\data\\access.log").map(x => {
      val splits = x.split(",")
      Access(splits(0).toLong, splits(1), splits(2).toLong)
    })
    

        val stream1 = env.addSource(new AccessSource)
        val stream2 = env.addSource(new AccessSource)

//1.数据类型是一样的 
        stream1.union(stream2).map(x=>{
          println("接收到的数据:" + x)
          x
        }).print()
  

    env.execute(this.getClass.getSimpleName)
  }

}

结果：不重要哈 结果不重要 
接收到的数据:Access(1577510963163,dongqiudi.com,656)
接收到的数据:Access(1577510963163,dongqiudi.com,523)
接收到的数据:Access(1577510963163,dongqiudi.com,340)
接收到的数据:Access(1577510963163,ruozedata.com,450)
接收到的数据:Access(1577510963163,ruozedata.com,790)
2> Access(1577510963163,ruozedata.com,450)
接收到的数据:Access(1577510963163,dongqiudi.com,211)
接收到的数据:Access(1577510963163,ruozedata.com,79)
接收到的数据:Access(1577510963163,dongqiudi.com,673)
4> Access(1577510963163,ruozedata.com,79)
6> Access(1577510963163,dongqiudi.com,211)

  /**
   * Creates a new DataStream by merging DataStream outputs of
   * the same type with each other. The DataStreams merged using this operator
   * will be transformed simultaneously.
   *
   */
  def union(dataStreams: DataStream[T]*): DataStream[T] =
    asScalaStream(stream.union(dataStreams.map(_.javaStream): _*))
union:  he same type with each other.  要求数据类型是一样的 两个流

但是生产上 要合并两个流 很少 两个流数据类型是一样的 

 /**
   * Creates a new ConnectedStreams by connecting
   * DataStream outputs of different type with each other. The
   * DataStreams connected using this operators can be used with CoFunctions.
   */
  def connect[T2](dataStream: DataStream[T2]): ConnectedStreams[T, T2] =
    asScalaStream(stream.connect(dataStream.javaStream))


两个流 数据类型不一样的： 使用 connect

两个流 数据类型不一样的：

def map[R: TypeInformation](fun1: IN1 => R, fun2: IN2 => R):


注意：
fun1: IN1 => R, fun2: IN2 => R

fun1: IN1 => R   第一个流  做的操作
 fun2: IN2 => R  第二个流 做的操作

object TranformationApp {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment


    val stream = env.readTextFile("C:\\IdeaProjects\\flink\\data\\access.log").map(x => {
      val splits = x.split(",")
      Access(splits(0).toLong, splits(1), splits(2).toLong)
    })

    val stream1 = env.addSource(new AccessSource)
    val stream2 = env.addSource(new AccessSource)
   
   val stream2New = stream2.map(x => ("J哥", x))
   
   stream1.connect(stream2New).map(x=>x,y=>y).print()

    env.execute(this.getClass.getSimpleName)
  }

}

结果是：
1> Access(1577511438837,zhibo8.cc,823)
8> Access(1577511438837,zhibo8.cc,379)
5> Access(1577511438837,zhibo8.cc,176)
5> Access(1577511438837,zhibo8.cc,414)
1> (J哥,Access(1577511438837,zhibo8.cc,553))
2> Access(1577511438837,ruozedata.com,775)
1> (J哥,Access(1577511438837,ruozedata.com,561))
6> Access(1577511438837,dongqiudi.com,657)
6> Access(1577511438837,zhibo8.cc,131)
6> (J哥,Access(1577511438837,dongqiudi.com,826))
7> Access(1577511438837,ruozedata.com,153)
7> (J哥,Access(1577511438837,ruozedata.com,98))
2> (J哥,Access(1577511438837,ruozedata.com,588))
2> (J哥,Access(1577511438837,ruozedata.com,351))
4> Access(1577511438837,ruozedata.com,24)
8> (J哥,Access(1577511438837,dongqiudi.com,812))
4> (J哥,Access(1577511438837,dongqiudi.com,686))
5> (J哥,Access(1577511438837,zhibo8.cc,825))
3> Access(1577511438837,dongqiudi.com,333)
3> (J哥,Access(1577511438837,zhibo8.cc,140))

connect: 
     Connects  two data streams retaining their types 
     数据结构可以不同
     two data streams  **
union：
       Union of two or more data streams
        数据结构要相同
        two or more data streams

Physical partitioning

class DoubleHappyPartitioner extends Partitioner[String]{
  
  override def partition(key: String, numPartitions: Int): Int = {
    println("partitions: " + numPartitions)
// 注意 scala 里面 不用使用 equals  直接 ==  即可
    if(key == "ruozedata.com"){
      0
    } else if(key == "dongqiudi.com"){
      1
    } else {
      2
    }

  }
}


注意：
Partitioner[String]

传进去的泛型 是 key的类型 

分区的前提 是 kv

/**
   * Partitions a tuple DataStream on the specified key fields using a custom partitioner.
   * This method takes the key position to partition on, and a partitioner that accepts the key
   * type.
   *
   * Note: This method works only on single field keys.
   */
  def partitionCustom[K: TypeInformation](partitioner: Partitioner[K], field: Int) : DataStream[T] =
    asScalaStream(stream.partitionCustom(partitioner, field))

field: Int  注意这个

object TranformationApp {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
        env.setParallelism(3)

        env.addSource(new AccessSource)
          .map(x=>(x.domain, x))
          .partitionCustom(new DoubleHappyPartitioner, 0)
          .map(x => {
            println("current thread id is: " + Thread.currentThread().getId + " , value is: " + x)
            x._2
          }).print()


    env.execute(this.getClass.getSimpleName)
  }

}

结果是：
partitions: 3
partitions: 3
partitions: 3
partitions: 3
partitions: 3
partitions: 3
partitions: 3
partitions: 3
partitions: 3
partitions: 3
current thread id is: 69 , value is: (zhibo8.cc,Access(1577512344598,zhibo8.cc,827))
current thread id is: 68 , value is: (ruozedata.com,Access(1577512344598,ruozedata.com,143))
1> Access(1577512344598,ruozedata.com,143)
3> Access(1577512344598,zhibo8.cc,827)
current thread id is: 68 , value is: (ruozedata.com,Access(1577512344598,ruozedata.com,672))
1> Access(1577512344598,ruozedata.com,672)
current thread id is: 70 , value is: (dongqiudi.com,Access(1577512344598,dongqiudi.com,450))
2> Access(1577512344598,dongqiudi.com,450)
current thread id is: 68 , value is: (ruozedata.com,Access(1577512344598,ruozedata.com,689))
current thread id is: 69 , value is: (zhibo8.cc,Access(1577512344598,zhibo8.cc,608))
current thread id is: 70 , value is: (dongqiudi.com,Access(1577512344598,dongqiudi.com,26))
3> Access(1577512344598,zhibo8.cc,608)
2> Access(1577512344598,dongqiudi.com,26)
1> Access(1577512344598,ruozedata.com,689)
current thread id is: 69 , value is: (zhibo8.cc,Access(1577512344598,zhibo8.cc,430))
current thread id is: 70 , value is: (dongqiudi.com,Access(1577512344598,dongqiudi.com,514))
3> Access(1577512344598,zhibo8.cc,430)
2> Access(1577512344598,dongqiudi.com,514)
current thread id is: 69 , value is: (zhibo8.cc,Access(1577512344598,zhibo8.cc,345))
3> Access(1577512344598,zhibo8.cc,345)

注意：线程
1-> ruozedata.com
2->dongqiudi.com
3>zhibo8.cc  
结果是没有问题的

这就是分区器的简单使用
让每个线程处理不同的数据

Flink01-double_happy

2020-01-05

介绍

在这里插入图片描述

Stateful Computations over Data Streams
    离线处理/批处理
        state watermark

    事件驱动：
    	Flink ：来一条处理一条
    	Spark：是微批次的  eg：3s     处理 3s的这些数据
    流批一体：

    Exactly-once state consistency
    Event-time processing   ***
    Sophisticated late data handling  延迟数据的处理（延迟数据是避免不了的）
    	（我现在知道的 离线处理 Spark是可以解决延迟数据的问题）

mvn archetype:generate                               \
-DarchetypeGroupId=org.apache.flink              \
-DarchetypeArtifactId=flink-quickstart-scala     \
-DarchetypeVersion=1.9.0



set up the batch execution environment
    ExecutionEnvironment.getExecutionEnvironment
Start with getting some data from the environment
then, transform the resulting DataSet[String] using operations
execute program


(x,...)


RichXXXFunction
生命周期函数
    open   初始化方法
    close  资源释放
    getRuntimeContext  拿到整个作业运行时上下文

Run Applications at any Scale
Flink is designed to run stateful streaming applications at any scale. Applications are parallelized into possibly thousands of tasks that are distributed and concurrently executed in a cluster. Therefore, an application can leverage virtually unlimited amounts of CPUs, main memory, disk and network IO. Moreover, Flink easily maintains very large application state. Its asynchronous and incremental checkpointing algorithm ensures minimal impact on processing latencies while guaranteeing exactly-once state consistency.

Users reported impressive scalability numbers for Flink applications running in their production environments, such as

applications processing multiple trillions of events per day,
applications maintaining multiple terabytes of state, and
applications running on thousands of cores

 maintaining multiple terabytes of state：
 注意：
 	state维护在hdfs 有什么好处和坏处呢？

Flink做的好的地方：
	1.处理时间  eventtime等 
	2.水印
	3.容错机制
	4.状态

pom.xml:

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <groupId>com.yt</groupId>
    <artifactId>Flink</artifactId>
    <version>1.0</version>
    <inceptionYear>2008</inceptionYear>
    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <flink.version>1.9.0</flink.version>
        <scala.binary.version>2.11</scala.binary.version>
        <scala.version>2.11.12</scala.version>
        <hadoop.version>2.7.2</hadoop.version>
    </properties>


    <repositories>
        <repository>
            <id>cloudera</id>
            <url>https://repository.cloudera.com/artifactory/cloudera-repos/</url>
        </repository>
    </repositories>


    <dependencies>
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-log4j12</artifactId>
            <version>1.7.7</version>
            <scope>runtime</scope>
        </dependency>

        <dependency>
            <groupId>log4j</groupId>
            <artifactId>log4j</artifactId>
            <version>1.2.17</version>
            <scope>runtime</scope>
        </dependency>
      <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-client</artifactId>
            <version>${hadoop.version}</version>
        </dependency>
        <!--Flink依赖-->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-scala_${scala.binary.version}</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-scala_${scala.binary.version}</artifactId>
            <version>${flink.version}</version>
        </dependency>

    </dependencies>

    <build>
        <sourceDirectory>src/main/scala</sourceDirectory>
        <testSourceDirectory>src/test/scala</testSourceDirectory>
        <plugins>
            <plugin>
                <groupId>org.scala-tools</groupId>
                <artifactId>maven-scala-plugin</artifactId>
                <executions>
                    <execution>
                        <goals>
                            <goal>compile</goal>
                            <goal>testCompile</goal>
                        </goals>
                    </execution>
                </executions>
                <configuration>
                    <scalaVersion>${scala.version}</scalaVersion>
                    <args>
                        <arg>-target:jvm-1.5</arg>
                    </args>
                </configuration>
            </plugin>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-eclipse-plugin</artifactId>
                <configuration>
                    <downloadSources>true</downloadSources>
                    <buildcommands>
                        <buildcommand>ch.epfl.lamp.sdt.core.scalabuilder</buildcommand>
                    </buildcommands>
                    <additionalProjectnatures>
                        <projectnature>ch.epfl.lamp.sdt.core.scalanature</projectnature>
                    </additionalProjectnatures>
                    <classpathContainers>
                        <classpathContainer>org.eclipse.jdt.launching.JRE_CONTAINER</classpathContainer>
                        <classpathContainer>ch.epfl.lamp.sdt.launching.SCALA_CONTAINER</classpathContainer>
                    </classpathContainers>
                </configuration>
            </plugin>
        </plugins>
    </build>
    <reporting>
        <plugins>
            <plugin>
                <groupId>org.scala-tools</groupId>
                <artifactId>maven-scala-plugin</artifactId>
                <configuration>
                    <scalaVersion>${scala.version}</scalaVersion>
                </configuration>
            </plugin>
        </plugins>
    </reporting>
</project>

批处理：
package com.sx.flink01

import org.apache.flink.api.scala.{DataSet, ExecutionEnvironment}
import org.apache.flink.api.scala._
object BatchJob {

  def main(args: Array[String]): Unit = {
    // 获取批处理上下文 <== SparkContext
    val env = ExecutionEnvironment.getExecutionEnvironment

    // 读取数据
    val text: DataSet[String] = env.readTextFile("C:\\IdeaProjects\\flink\\data\\data.txt")

    val value: DataSet[(String, Int)] = text.flatMap(_.toLowerCase.split(","))
      .filter(_.nonEmpty)
      .map((_, 1))

    // transformation
    val result = value.groupBy(0) // 0表示单词
      .sum(1)

    // sink  output
    result.print()
  }
}


结果：
(double_happy,1)
(kite,1)
(25,2)
(kairis,1)
(32,1)

[double_happy@hadoop101 ~]$ nc -lk 7777
doublehappy,1
doublehappy,2
doublehappy,3
-------

package com.sx.flink01

import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.api.scala._

object StreamingJob {

  def main(args: Array[String]): Unit = {
    // set up the streaming execution environment
    val env = StreamExecutionEnvironment.getExecutionEnvironment
  
    // 接收数据
    val text: DataStream[String] = env.socketTextStream("hadoop101", 7777)

    // transformation
        val result = text.flatMap(_.toLowerCase.split(","))
          .filter(_.nonEmpty)
          .map((_, 1))
          .keyBy(0).sum(1).print()

    env.execute(this.getClass.getSimpleName)

  }

}

结果是：
2> (2,1)
4> (doublehappy,1)
4> (1,1)
4> (doublehappy,2)
4> (doublehappy,3)
3> (3,1)

[double_happy@hadoop101 ~]$ nc -lk 7777
double,double,double
happy,happy
------------------------

package com.sx.flink01

import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.api.scala._

object StreamingJob {

  def main(args: Array[String]): Unit = {

    // set up the streaming execution environment
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    // 接收数据
    val text: DataStream[String] = env.socketTextStream("hadoop101", 7777)

    // transformation
        val result = text.flatMap(_.toLowerCase.split(","))
          .filter(_.nonEmpty)
          .map((_, 1))
          .keyBy(0).sum(1).print("double_happy")
          .setParallelism(2)
    env.execute(this.getClass.getSimpleName)

  }
}

结果：
double_happy:1> (double,2)
double_happy:2> (double,1)
double_happy:2> (double,3)
double_happy:1> (happy,1)
double_happy:2> (happy,2)


并行度为1：结果
double_happy> (double,1)
double_happy> (double,2)
double_happy> (double,3)
double_happy> (happy,1)
double_happy> (happy,2)

使用字段表达式

package com.sx.flink01

import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.api.scala._

object StreamingJob {

  def main(args: Array[String]): Unit = {

    // set up the streaming execution environment
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    // 接收数据
    val text: DataStream[String] = env.socketTextStream("hadoop101", 7777)

    // transformation
//        val result = text.flatMap(_.toLowerCase.split(","))
//          .filter(_.nonEmpty)
//          .map((_, 1))
//          .keyBy(0).sum(1).print("double_happy")
//          .setParallelism(1)

    val result = text.flatMap(_.toLowerCase.split(","))
      .filter(_.nonEmpty)
      .map(x => WC(x, 1))
      .keyBy(_.word).sum("count").print("double_happy")
      .setParallelism(1)
//
    // execute program
    env.execute(this.getClass.getSimpleName)

  }

}

case class WC(word: String, count: Int)

在这里插入图片描述

依赖

在这里插入图片描述

At the bare minimum, the application depends on the Flink APIs. Many applications depend in addition on certain connector libraries (like Kafka, Cassandra, etc.)

注意：
	connector libraries

Specifying Transformation Functions

package com.sx.bean

object Domain {
  case class Access(time:Long, domain:String, traffic:Long)

}

package com.sx.flink01

import com.sx.bean.Domain.Access
import org.apache.flink.api.common.functions.{FilterFunction, RichMapFunction, RuntimeContext}
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.api.scala._

object SpecifyingTransformationFunctionsApp {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    val stream = env.readTextFile("C:\\IdeaProjects\\flink\\data\\access.log")
    val accessStream = stream.map(x => {
      val splits = x.split(",")
      Access(splits(0).toLong, splits(1), splits(2).toLong)
    })

    accessStream.print()
    env.execute(this.getClass.getSimpleName)
  }
}

结果：
Access(201912120010,ruozedata.com,2000)
Access(201912120010,dongqiudi.com,6000)
Access(201912120010,zhibo8.com,5000)
Access(201912120010,ruozedata.com,4000)
Access(201912120010,dongqiudi.com,1000)

需求1
	过滤traffic > 4000

方式1：Lambda Functions
object SpecifyingTransformationFunctionsApp {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    val stream = env.readTextFile("C:\\IdeaProjects\\flink\\data\\access.log")
    val accessStream = stream.map(x => {
      val splits = x.split(",")
      Access(splits(0).toLong, splits(1), splits(2).toLong)
    })
     // 过滤traffic > 4000
     accessStream.filter(_.traffic > 4000).print()
    env.execute(this.getClass.getSimpleName)
  }
}


结果：
Access(201912120010,dongqiudi.com,6000)
Access(201912120010,zhibo8.com,5000)

方式二：
package com.sx.flink01

import com.sx.bean.Domain.Access
import org.apache.flink.api.common.functions.{FilterFunction, RichMapFunction, RuntimeContext}
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.api.scala._

object SpecifyingTransformationFunctionsApp {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    val stream = env.readTextFile("C:\\IdeaProjects\\flink\\data\\access.log")
    val accessStream = stream.map(x => {
      val splits = x.split(",")
      Access(splits(0).toLong, splits(1), splits(2).toLong)
    })
    accessStream.filter(new DoubleHappyFilter02(5000)).print()
    
    //    accessStream.filter(new FilterFunction[Access] {
    //      override def filter(value: Access): Boolean = value.traffic>4000
    //    }).print()     //匿名内部类写法   别这么写  太难看了

    env.execute(this.getClass.getSimpleName)
  }
}

// XXXFunction   RichXXXFunction
class DoubleHappyFilter extends FilterFunction[Access] {
  override def filter(value: Access): Boolean = value.traffic > 4000
}

class DoubleHappyFilter02(traffic: Long) extends FilterFunction[Access] {
  override def filter(value: Access): Boolean = value.traffic > traffic
}


结果：
Access(201912120010,dongqiudi.com,6000)

在这里插入图片描述
这里面的生命周期函数

RichXXXFunction
生命周期函数
    open   初始化方法
    close  资源释放
    getRuntimeContext  拿到整个作业运行时上下文

方式3：Rich functions

package com.sx.flink01

import com.sx.bean.Domain.Access
import org.apache.flink.api.common.functions.{FilterFunction, RichMapFunction, RuntimeContext}
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.api.scala._

object SpecifyingTransformationFunctionsApp {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(2)
    
    val stream = env.readTextFile("C:\\IdeaProjects\\flink\\data\\access.log")
     val accessStream = stream.map(new DoubleHappyMap)

     // 过滤traffic > 4000
     accessStream.filter(_.traffic > 4000).print()
    env.execute(this.getClass.getSimpleName)
  }
}

// XXXFunction   RichXXXFunction
class DoubleHappyFilter extends FilterFunction[Access] {
  override def filter(value: Access): Boolean = value.traffic > 4000
}

class DoubleHappyFilter02(traffic: Long) extends FilterFunction[Access] {
  override def filter(value: Access): Boolean = value.traffic > traffic
}

class DoubleHappyMap extends RichMapFunction[String, Access] {
  override def map(value: String): Access = {
    val splits = value.split(",")
    Access(splits(0).toLong, splits(1), splits(2).toLong)
  }

  override def open(parameters: Configuration): Unit = {
    super.open(parameters)
    println("~~~~~~~~~~~open~~~~~~~")
  }
  override def close(): Unit = {
    super.close()
  }
  override def getRuntimeContext: RuntimeContext = {
    super.getRuntimeContext
  }
}


结果：  并行度为2的时候  
~~~~~~~~~~~open~~~~~~~
~~~~~~~~~~~open~~~~~~~
1> Access(201912120010,dongqiudi.com,6000)
1> Access(201912120010,zhibo8.com,5000)

这块要注意 ：MySQL获取connection 放在open方法里 
还要注意  不同的并行读  调用了几次 

结果：并行度为1的时候
~~~~~~~~~~~open~~~~~~~
Access(201912120010,dongqiudi.com,6000)
Access(201912120010,zhibo8.com,5000)

tmp

2019-11-20

Azkaban调度--double_happy

2019-01-05

安装Azkaban的坑

当你使用单机模式运行azkaban时候：

2019/09/23 22:06:53.626 +0800 ERROR [AzkabanWebServer] [Azkaban] Failed to start single server. Shutting down.
java.io.IOException: Cannot find 'database.properties' file in /home/double_happy/app/azkaban-solo-server/bin/sql/database.properties
        at azkaban.database.AzkabanDatabaseSetup.loadDBProps(AzkabanDatabaseSetup.java:178)
        at azkaban.database.AzkabanDatabaseSetup.loadTableInfo(AzkabanDatabaseSetup.java:102)
        at azkaban.database.AzkabanDatabaseUpdater.runDatabaseUpdater(AzkabanDatabaseUpdater.java:82)
        at azkaban.soloserver.AzkabanSingleServer.start(AzkabanSingleServer.java:93)
        at azkaban.soloserver.AzkabanSingleServer.main(AzkabanSingleServer.java:58)

解决方式：

必须使用bin/azkaban-solo-start.sh

调度解决多个作业之间的依赖关系

nodes:
  - name: platform_stat
    type: command
    dependsOn:
      - etl
    config:
      command: sh /home/double_happy/ruozedata/project/hadoop-project/shell/platform_stat.sh 20190921

  - name: etl
    type: command
    config:
      command: sh /home/double_happy/ruozedata/project/hadoop-project/shell/etl.sh 20190921

在这里插入图片描述

但是又一个问题呀：
	shell 里的参数我可以不可以在 azkaban上面通过 parameters 传进来呢？
	待续。。。。。。。

Zookeeper基本使用与监控(Curator)

2019-01-04

1.Zookeeper基本使用

(0)基本了解

ZK核心术语
    ZK的数据模型： 树形结构
        /home/hadoop/app/ruozedata.txt
    ZK每个节点znode，有一个唯一的路径标识
    znode
        临时的 ephemeral  临时znode下面不能有子节点
            普通的
            sequential
        永久的 persistent
            普通的
            sequential  顺序编号目录节点
    每个znode节点有各自的版本号
    每个节点数据发生了变化，该节点的版本会加
    ZK节点存储的数据量不宜过大，几K  *****
    znode可以设置访问权限
    znode可以设置watcher：当节点数据发生变化时，可以通过监视器获取

(1)客户端命令行操作

 命令基本语法						功能描述
help										显示所有操作命令
ls path [watch]						使用 ls 命令来查看当前znode中所包含的内容
ls2 path [watch]					查看当前节点数据并能看到更新次数等数据
create									普通创建
											-s  含有序列
											-e  临时（重启或者超时消失）
get path [watch]					获得节点的值
set										设置节点的具体值
stat										查看节点状态
delete									删除节点
rmr										递归删除节点

上面带[watch]  的是可以监控的  但是watcher zookeeper自身的api和命令都是一次性的 
也就是用过了不能再用了（你懂的 ）
解决办法 可以使用 curator 作为zkClient 来监控zookeeper的变化信息****

eg：一些监控的种类
watcher
    zk中的watcher是一次性(*****)的
    NodeCreated
    NodeDataChanged
    NodeDeleted
    NodeChildrenChanged
    
四字命令 了解一下

(2)CuratorApi
Curator是Netflix公司开源的一套zookeeper客户端框架，解决了很多Zookeeper客户端非常底层的细节开发工作，包括连接重连、反复注册Watcher和NodeExistsException异常等等。

版本选择：

这里我们的zookeeper的版本是3.4.x，所以api开发的时候 maven不同版本会用不同
https://curator.apache.org/zk-compatibility.html      curator官网

curator api 官网：https://curator.apache.org/curator-framework/index.html

可以去官网上去学习，可以结合zookeeper api 来学习 curator api （突然手里来活了写不了了）
这里可以借鉴：
http://www.throwable.club/2018/12/16/zookeeper-curator-usage/#%E5%89%8D%E6%8F%90
英文不好的小伙伴，这篇文章很好的介绍Curator 使用。

还有我的小组长的博客：
https://liverrrr.fun/archives/quickstart_zookeeper
可以很好的学习。