2019-12-09

All Blogs►Microsoft Azure►Database►MySQL►Linux

Azure Database Migration Service — 数据库迁移到 Microsoft Azure 平台的好帮手

1. 前言

在复杂多云的环境下，数据库迁移成为了一个既常见又头疼的问题，所以又快又好并保证数据一致性的数据库迁移服务就显得非常必要了。基于这点的考量，Microsoft Azure 也发布了自己的数据迁移服务 Azure Database Migration Service 来帮助想把各类数据库迁移到 Azure 的客户解决数据库迁移的后顾之忧。

2. Azure Database Migration Service 介绍

Azure Migration Service 是完全托管的服务，旨在实现从多个数据库源无缝迁移到 Azure 并只让应用程序的停机时间最短。该服务集成了一些现有工具和服务的功能从而为客户提供高度可用的综合解决方案，例如使用数据迁移助手生成评估报告，这些报告可以提供建议并能够指导执行迁移之前完成所需的更改。该过程利用了 Microsoft 的最佳实践，因此客户可以在启动迁移项目后便高枕无忧。数据库迁移的过程原理如下图所示：

从上至下来看，在 Pre-migration 阶段评估工作量并做 Schema 的映射，继而开始做接近同步的异步 Migration，最后可以通过 Cutover 来做应用程序的迁移。除此之外，为了让大家能够更好的学习该服务，Microsoft 也非常良心的提供了学习的素材，不仅包括官方文档，还有很多视频材料可以参考。为了便捷，本文以下内容都使用 Azure Database Migration Service 的简称 Azure DMS，好了话不多说，我们直接来实战操作吧！

3. Aliyun RDS MySQL 迁移至 Azure Mooncake MySQL PaaS 验证

3.1 实验场景以及前期准备

为了尽可能的模拟真实生产，本文模拟从中国 Aliyun RDS MySQL 到 Azure Mooncake MySQL PaaS，Mooncake 为中国区 Azure 的代号。具体的 MySQL 版本为5.7，具体信息如下：

1	Ⅰ. Aliyun RDS MySQL -> Azure Mooncake MySQL PaaS (中国区)

在进行实验之前，需要正确安装 Terraform、MySQL Client 等软件包，详细步骤请参考相应文档，本文不进行赘述。具体的客户端连接 MySQL Server 的操作过程在 WSL2-Ubuntu-18.04 以及 Windows 10 的 Navicat 12 客户端上进行。

3.2 Aliyun RDS MySQL 迁移到 Azure Mooncake MySQL PaaS (中国区)

3.2.1 Azure Mooncake 资源创建初始化

首先，通过 Terraform 自动化部署 MySQL Generel Purpose 2core 实例，需要注意的是在中国区 Azure 创建资源的时候需要指定 environment 为 china，具体的 .tf 文件内容如下：

# Configure the Microsoft Azure Provider
provider "azurerm" {
    subscription_id = "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx"
    client_id       = "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx"
    client_secret   = "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx"
    tenant_id       = "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx"
    environment     = "china"
}

# Create a resource group if it doesn’t exist
resource "azurerm_resource_group" "myterraformgroup" {
    name     = "azcndmsrg0001"
    location = "chinaeast2"
    tags = {
        environment = "Azure Terraform Automation"
    }
}

# Create virtual network
resource "azurerm_virtual_network" "myterraformnetwork" {
    name                = "azcndmsvnet0001"
    address_space       = ["10.91.0.0/16"]
    location            = "chinaeast2"
    resource_group_name = "${azurerm_resource_group.myterraformgroup.name}"

    tags = {
        environment = "Azure Terraform Automation"
    }
}

# Create Network Security Group and rule
resource "azurerm_network_security_group" "publicnsg" {
    name                = "public-nsg"
    location            = "chinaeast2"
    resource_group_name = "${azurerm_resource_group.myterraformgroup.name}"
    
    security_rule {
        name                       = "SSH"
        priority                   = 1001
        direction                  = "Inbound"
        access                     = "Allow"
        protocol                   = "Tcp"
        source_port_range          = "*"
        destination_port_range     = "22"
        source_address_prefix      = "*"
        destination_address_prefix = "*"
    }
    
        security_rule {
        name                       = "RDP"
        priority                   = 1002
        direction                  = "Inbound"
        access                     = "Allow"
        protocol                   = "Tcp"
        source_port_range          = "*"
        destination_port_range     = "3389"
        source_address_prefix      = "*"
        destination_address_prefix = "*"
    }

    tags = {
        environment = "Azure Terraform Automation"
    }
}
resource "azurerm_network_security_group" "opnsg" {
    name                = "op-nsg"
    location            = "chinaeast2"
    resource_group_name = "${azurerm_resource_group.myterraformgroup.name}"
    
    security_rule {
        name                       = "SSH"
        priority                   = 1001
        direction                  = "Inbound"
        access                     = "Allow"
        protocol                   = "Tcp"
        source_port_range          = "*"
        destination_port_range     = "22"
        source_address_prefix      = "*"
        destination_address_prefix = "*"
    }
    
        security_rule {
        name                       = "RDP"
        priority                   = 1002
        direction                  = "Inbound"
        access                     = "Allow"
        protocol                   = "Tcp"
        source_port_range          = "*"
        destination_port_range     = "3389"
        source_address_prefix      = "*"
        destination_address_prefix = "*"
    }

    tags = {
        environment = "Azure Terraform Automation"
    }

# Create subnet
resource "azurerm_subnet" "publicsubnet" {
    name                 = "public0001"
    resource_group_name  = "${azurerm_resource_group.myterraformgroup.name}"
    virtual_network_name = "${azurerm_virtual_network.myterraformnetwork.name}"
    address_prefix       = "10.91.0.0/23"
    network_security_group_id = "${azurerm_network_security_group.publicnsg.id}"
}
resource "azurerm_subnet" "opsubnet" {
    name                 = "op0001"
    resource_group_name  = "${azurerm_resource_group.myterraformgroup.name}"
    virtual_network_name = "${azurerm_virtual_network.myterraformnetwork.name}"
    address_prefix       = "10.91.2.0/23"
    network_security_group_id = "${azurerm_network_security_group.opnsg.id}"
}

# Create Azure Database for MySQL 
resource "azurerm_mysql_server" "mysql0001" {
  name                = "mysql0001"
  location            = "${azurerm_resource_group.myterraformgroup.location}"
  resource_group_name = "${azurerm_resource_group.myterraformgroup.name}"

  sku {
    name     = "GP_Gen5_2"
    capacity = 2
    tier     = "GeneralPurpose"
    family   = "Gen5"
  }

  storage_profile {
    storage_mb            = 20480
    backup_retention_days = 7
    geo_redundant_backup  = "Disabled"
  }

  administrator_login          = "msadmin"
  administrator_login_password = "Microsoft2019!"
  version                      = "5.7"
  ssl_enforcement              = "Enabled"
  
}

整个部署过程大概10分钟左右，整个初始化过程会创建资源组、Vnet、NSG、Azure MySQL PaaS 实例等。信息如下：

需要注意的是，Azure MySQL PaaS 没有启用 SSL。

3.2.2 创建 Aliyun RDS MySQL 实例

创建 Aliyun RDS MySQL 实例的过程不予赘述，通过 Portal 一步一步点击即可，本文主要阐述功能性验证，所以部署的实例类型为 MySQL 5.7 基础版 1Core2G 的 mysql.n2.small.1 实例，并启用公网地址。

创建好之后的需要设置一下白名单以便能够使 Azure DMS 和 Aliyun RDS MySQL 通过公网调用，Azure China Service IP Range 可以在这里下载，然后把这些网段加到白名单中即可，避免了直接写0.0.0.0/0。

3.2.3 导入示例数据库到 Aliyun RDS MySQL 实例

本文采用 MySQL 官方提供的 employees 数据库，可以参考具体的官网地址和 Github地址。该实例数据库包含了300,000 员工及收入信息，数据大概100MB+，数据量已经足够进行测试。employees 库中，表之间的调用关系如下图所示：

下面我们开始进行实验，首先通过 Linux MySQL Client 连接至 Aliyun RDS MySQL 实例安装并导入数据库：

1
2
3

$ git clone https://github.com/datacharmer/test_db.git
$ cd test_db/
$ mysql -u'msadmin' -p'Microsoft2019!' -h'rm-uf626yx145z16o277so.mysql.rds.aliyuncs.com' < employees.sql

通过 SQL 查看具体的表容量：

mysql> select table_schema as '数据库', sum(table_rows) as '记录数', sum(truncate(data_length/1024/1024, 2)) as '数据容量(MB)', sum(truncate(index_length/1024/1024, 2)) as '索引容量(MB)' from information_schema.tables group by table_schema order by sum(data_length) desc, sum(index_length) desc;
+--------------------+-----------+------------------+------------------+
| 数据库             | 记录数    | 数据容量(MB)     | 索引容量(MB)     |
+--------------------+-----------+------------------+------------------+
| employees          |   3911631 |           141.22 |             5.53 |
| mysql              |    126112 |             5.41 |             0.07 |
| information_schema |      NULL |             0.10 |             0.00 |
| sys                |         0 |             0.01 |             0.00 |
| performance_schema |     11701 |             0.00 |             0.00 |
+--------------------+-----------+------------------+------------------+
5 rows in set (0.22 sec)

安装完可以进行验证：

$ mysql -u'msadmin' -p'Microsoft2019!' -h'rm-uf626yx145z16o277so.mysql.rds.aliyuncs.com' -t < test_employees_md5.sql 
+----------------------+
| INFO                 |
+----------------------+
| TESTING INSTALLATION |
+----------------------+
+--------------+------------------+----------------------------------+
| table_name   | expected_records | expected_crc                     |
+--------------+------------------+----------------------------------+
| departments  |                9 | d1af5e170d2d1591d776d5638d71fc5f |
| dept_emp     |           331603 | ccf6fe516f990bdaa49713fc478701b7 |
| dept_manager |               24 | 8720e2f0853ac9096b689c14664f847e |
| employees    |           300024 | 4ec56ab5ba37218d187cf6ab09ce1aa1 |
| salaries     |          2844047 | fd220654e95aea1b169624ffe3fca934 |
| titles       |           443308 | bfa016c472df68e70a03facafa1bc0a8 |
+--------------+------------------+----------------------------------+
+--------------+------------------+----------------------------------+
| table_name   | found_records    | found_crc                        |
+--------------+------------------+----------------------------------+
| departments  |                9 | d1af5e170d2d1591d776d5638d71fc5f |
| dept_emp     |           331603 | ccf6fe516f990bdaa49713fc478701b7 |
| dept_manager |               24 | 8720e2f0853ac9096b689c14664f847e |
| employees    |           300024 | 4ec56ab5ba37218d187cf6ab09ce1aa1 |
| salaries     |          2844047 | fd220654e95aea1b169624ffe3fca934 |
| titles       |           443308 | bfa016c472df68e70a03facafa1bc0a8 |
+--------------+------------------+----------------------------------+
+--------------+---------------+-----------+
| table_name   | records_match | crc_match |
+--------------+---------------+-----------+
| departments  | OK            | ok        |
| dept_emp     | OK            | ok        |
| dept_manager | OK            | ok        |
| employees    | OK            | ok        |
| salaries     | OK            | ok        |
| titles       | OK            | ok        |
+--------------+---------------+-----------+
+------------------+
| computation_time |
+------------------+
| 00:00:21         |
+------------------+
+---------+--------+
| summary | result |
+---------+--------+
| CRC     | OK     |
| count   | OK     |
+---------+--------+

3.2.4 创建 Azure DMS 实例

登陆到 Azure China Portal，选择 All services -> Azure Database Migration Services -> 点击 Add：

注意，Azure DMS 有 Standard 和 Premium 两个 SKU 可以选择，具体的性能指标和价格可以看这里。本实验选择 Premium 的主要原因是 Premium SKU 支持 Online 的迁移。

3.2.5 在目标 Azure MySQL PaaS 上创建 Schema

通过 mysqldump 配合 –no-data 从 Aliyun RDS MySQL 导出 Schema SQL 并导入到 Azure MySQL PaaS 中：

$ mysqldump -u'msadmin' -p'Microsoft2019!' -h'rm-uf626yx145z16o277so.mysql.rds.aliyuncs.com' --databases 'employees' --no-data > sourcedata.sql
$ sed -e 's/DEFINER[ ]*=[ ]*[^*]*\*/\*/ ' sourcedata.sql > sourcedatanew.sql
$ awk '{ if (index($0,"GTID_PURGED")) { getline; while (length($0) > 0) { getline; } } else { print $0 } }' sourcedatanew.sql | grep -iv 'set @@' > sourcedata.sql
$ mysql -u'msadmin@mysql0001' -p'Microsoft2019!' -h'mysql0001.mysql.database.chinacloudapi.cn' < sourcedata.sql

导入之后需要删除 Azure MySQL PaaS employees 库里的 Foreign Key(s)，查询外键的 SQL 如下：

$ mysql> SELECT SchemaName, GROUP_CONCAT(DropQuery SEPARATOR ';\n') as DropQuery, GROUP_CONCAT(AddQuery SEPARATOR ';\n') as AddQuery
    -> FROM
    -> (SELECT 
    -> KCU.REFERENCED_TABLE_SCHEMA as SchemaName, KCU.TABLE_NAME, KCU.COLUMN_NAME,
    -> CONCAT('ALTER TABLE ', KCU.TABLE_NAME, ' DROP FOREIGN KEY ', KCU.CONSTRAINT_NAME) AS DropQuery,
    ->     CONCAT('ALTER TABLE ', KCU.TABLE_NAME, ' ADD CONSTRAINT ', KCU.CONSTRAINT_NAME, ' FOREIGN KEY (`', KCU.COLUMN_NAME, '`) REFERENCES `', KCU.REFERENCED_TABLE_NAME, '` (`', KCU.REFERENCED_COLUMN_NAME, '`) ON UPDATE ',RC.UPDATE_RULE, ' ON DELETE ',RC.DELETE_RULE) AS AddQuery
    -> FROM INFORMATION_SCHEMA.KEY_COLUMN_USAGE KCU, information_schema.REFERENTIAL_CONSTRAINTS RC
    -> WHERE
    ->   KCU.CONSTRAINT_NAME = RC.CONSTRAINT_NAME
    ->   AND KCU.REFERENCED_TABLE_SCHEMA = RC.UNIQUE_CONSTRAINT_SCHEMA) Queries
    ->   GROUP BY SchemaName;

选择出结果中关于 Drop 的语句并执行：

mysql> use employees;
mysql> ALTER TABLE dept_emp DROP FOREIGN KEY dept_emp_ibfk_1;
mysql> ALTER TABLE dept_emp DROP FOREIGN KEY dept_emp_ibfk_2;
mysql> ALTER TABLE dept_manager DROP FOREIGN KEY dept_manager_ibfk_1;
mysql> ALTER TABLE dept_manager DROP FOREIGN KEY dept_manager_ibfk_2;
mysql> ALTER TABLE salaries DROP FOREIGN KEY salaries_ibfk_1;
mysql> ALTER TABLE titles DROP FOREIGN KEY titles_ibfk_1;

至此，Schema 在 Azure MySQL PaaS 上迁移完毕。

3.2.5 创建 Migration Project 并做 Online Migration

创建过程按照 Migration Wizards 一步一步操作即可，注意需要在 Azure MySQL PaaS 上配置允许 Azure DMS 的访问，Source Database 类型选择 MySQL 或者 Amazon RDS MySQL 都可以。

配置好之后，直接 Run Migration 就可以开始做迁移了，迁移结束后的状态如图所示。值得注意的是，由于选择的迁移类型是 Online 的，所以迁移结束后，DMS 还处于 Running 状态。如果此时需要停止，可以进行 Cutover。

离线迁移做完，现在我们模拟在 Aliyun RDS MySQL 的实时写入并在 Azure MySQL PaaS 上查看异步同步的数据来模拟 Online 迁移。具体写入 employees 库的 employees 表，插入主键 emp_no 范围在 10 ~ 153 这 144 条记录，为了避免冲突，可以考虑先在插入之前用 SQL Query 确认一下是否有插入的字段。插入脚本如下：

$ for time in $(seq 10 153)
> do
> mysql -u'msadmin' -p'Microsoft2019!' -h'rm-uf626yx145z16o277so.mysql.rds.aliyuncs.com' -e "INSERT INTO employees.employees (emp_no, first_name, last_name) VALUES ('$time', 'Xinsheng$time', 'Wang$time');"
> sleep 1
> done

为了方便更直观的查看结果，该结果使用 Navicat 12 连接数据库，具体插入同步结果如下：
Aliyun RDS MySQL

Azure MySQL PaaS

144条记录同步确认

通过 Azure DMS 可以看出来异步同步的记录数量，同时查询数据库也可以做双重验证。

除此之外，Azure DMS 也能够根据写入的速度、频率及数据量大小来评估出来应用程序在做 Cutover 的时候的停机时间，这也是非常的人性化了。

4. 总结

至此，跨云平台的 MySQL PaaS 迁移已经完成了，通过 Azure DMS 可以做到 Online 的数据库迁移至 Azure 并且能够给出校验结果。但是本文还有一些可以优化的场景，比如说 Azure DMS 的迁移可以通过 VPN 链路，VPN 通过两朵云的 VPN Gateway 打通即可，亲测可行，篇幅有限就不放在这篇博客了，留给大家自己测试吧。

2019-11-17

All Blogs►Microsoft Azure►Kubernetes►Bigdata►Flink►Linux

新一代大数据计算框架 Flink 与 Microsoft Azure Kubernetes 集成来做流批计算

1. 前言

上一篇，我们验证测试了 Apache Spark 和 Azure Kubernetes Service 的集成，包括通过 Cluster Mode 依托 AKS 集群运行 Spark 作业以及和 Azure Blob Storage 集成来做计算存储的解耦分离。和传统的资源调度框架 Hadoop Yarn 相比，使用 Azure Kubernetes Service 提高了部署速度和效率，提高灵活性并一定程度降低了运维管理成本。Spark 在强大的社区支撑下，技术的成熟度是在几大大数据计算框架中最高也是最全面的。但是除了 Spark，在大数据计算框架中还有一颗冉冉升起的新星，It is Apache Flink。本文我们来聊聊 Apache Flink 以及如何与 Azure Kubernetes Service 集成来做流批计算。

2. Apahce Flink 介绍

Apache Flink 是一个分布式大数据处理引擎，可对有限数据流和无限数据流进行有状态或无状态的计算，能够部署在各种集群环境，对各种规模大小的数据进行快速计算。

（from: https://flink.apache.org/)

Flink 的代码主要是由 Java 语言实现的，部分代码是 Scala，提供主流的 Java 及 Scala API，并且从 1.0 版本开始也提供 Python API 即 Pyflink。对 Flink 而言，其所要处理的主要场景就是 Streaming 流数据，Batch 批数据只是流数据的一个时间维度上的特例而已，所以 Flink 是一款真正的流批统一的计算引擎，这种设计思路对于某些业务场景可以提高代码的复用率，所以这也是 Flink 被很多大厂接受并广泛使用的原因之一。下面我们来看看 Flink 的内部架构：

从下至上来看：
部署： Flink 支持本地运行（IDE 中直接运行程序）、能在独立集群（Standalone 模式）或者在被 YARN、Mesos、K8s 管理的集群上运行，也能部署在云上。
运行：Flink 的核心是分布式流式数据引擎，意味着数据以一次一个事件的形式被处理。
API：DataStream、DataSet、Table、SQL API。
扩展库：Flink 还包括用于 CEP（复杂事件处理）、机器学习、图形处理等场景。

本文仅调研 Flink on Azure Kubernetes 并做流批场景的技术可行性验证，对于其他部署模式不展开讨论，有兴趣的同学可以自行研究。

3. Flink on AKS 实现

本实验需要正确安装 Azure Cli、Terraform，Kubectl、Docker 等软件包，详细步骤请参考相应文档，本文不进行赘述。具体的操作过程在 AKS Vnet 的一台内网服务器上进行，该服务器作为 Flink 集群的跳板机使用。

3.1 创建 AKS 集群、ACR 及跳板机

为了提高效率，本文直接使用 Terraform 来做自动化一键部署，Terraform 的介绍不做赘述，同学们自行搜索吧。废话少说，直接上 Code，具体 Terraform Azure 的用法可以参考这里。本文按照 Terraform 的最佳实践结构设计，由于 variables.tf 包含一些个人 Azure 账号信息，所以本文只展现主要的 azure-aks.tf 文件，内容如下：

# Configure the Microsoft Azure Provider
provider "azurerm" {
    subscription_id = "${var.subscription_id}"
    client_id       = "${var.client_id}"
    client_secret   = "${var.client_secret}"
    tenant_id       = "${var.tenant_id}"
}

# Create a resource group if it doesn’t exist
resource "azurerm_resource_group" "myterraformgroup" {
    name     = "${var.resource_group_name}"
    location = "${var.location}"
    tags = {
        environment = "Azure Terraform Kubernetes Service"
    }
}

# Create virtual network
resource "azurerm_virtual_network" "myterraformnetwork" {
    name                = "${var.vnet_name}"
    address_space       = "${var.vnet_address_space}"
    location            = "${azurerm_resource_group.myterraformgroup.location}"
    resource_group_name = "${azurerm_resource_group.myterraformgroup.name}"

    tags = {
        environment = "Azure Terraform Kubernetes Service"
    }
}

# Create Network Security Group and rule
resource "azurerm_network_security_group" "public_nsg" {
    name                = "${var.public_nsg}"
    location            = "${azurerm_resource_group.myterraformgroup.location}"
    resource_group_name = "${azurerm_resource_group.myterraformgroup.name}"
    
    security_rule {
        name                       = "SSH"
        priority                   = 1001
        direction                  = "Inbound"
        access                     = "Allow"
        protocol                   = "Tcp"
        source_port_range          = "*"
        destination_port_range     = "22"
        source_address_prefix      = "*"
        destination_address_prefix = "*"
    }

    tags = {
        environment = "Azure Terraform Kubernetes Service"
    }
}
resource "azurerm_network_security_group" "op_nsg" {
    name                = "${var.op_nsg}"
    location            = "${azurerm_resource_group.myterraformgroup.location}"
    resource_group_name = "${azurerm_resource_group.myterraformgroup.name}"
    
    security_rule {
        name                       = "SSH"
        priority                   = 1001
        direction                  = "Inbound"
        access                     = "Allow"
        protocol                   = "Tcp"
        source_port_range          = "*"
        destination_port_range     = "22"
        source_address_prefix      = "*"
        destination_address_prefix = "*"
    }

    tags = {
        environment = "Azure Terraform Kubernetes Service"
    }
}
resource "azurerm_network_security_group" "aks_nsg" {
    name                = "${var.aks_nsg}"
    location            = "${azurerm_resource_group.myterraformgroup.location}"
    resource_group_name = "${azurerm_resource_group.myterraformgroup.name}"
    
    security_rule {
        name                       = "SSH"
        priority                   = 1001
        direction                  = "Inbound"
        access                     = "Allow"
        protocol                   = "Tcp"
        source_port_range          = "*"
        destination_port_range     = "22"
        source_address_prefix      = "*"
        destination_address_prefix = "*"
    }
    security_rule {
        name                       = "AKSWebUI"
        priority                   = 1002
        direction                  = "Inbound"
        access                     = "Allow"
        protocol                   = "Tcp"
        source_port_range          = "*"
        destination_port_range     = "8081"
        source_address_prefix      = "*"
        destination_address_prefix = "*"
    }

    tags = {
        environment = "Azure Terraform Kubernetes Service"
    }
}

# Create subnet
resource "azurerm_subnet" "public_subnet" {
    name                 = "${var.public_subnet}"
    resource_group_name  = "${azurerm_resource_group.myterraformgroup.name}"
    virtual_network_name = "${azurerm_virtual_network.myterraformnetwork.name}"
    address_prefix       = "${var.public_subnet_address_prefix}"
    network_security_group_id = "${azurerm_network_security_group.public_nsg.id}"
}
resource "azurerm_subnet" "op_subnet" {
    name                 = "${var.op_subnet}"
    resource_group_name  = "${azurerm_resource_group.myterraformgroup.name}"
    virtual_network_name = "${azurerm_virtual_network.myterraformnetwork.name}"
    address_prefix       = "${var.op_subnet_address_prefix}"
    network_security_group_id = "${azurerm_network_security_group.public_nsg.id}"
}
resource "azurerm_subnet" "aks_subnet" {
    name                 = "${var.aks_subnet}"
    resource_group_name  = "${azurerm_resource_group.myterraformgroup.name}"
    virtual_network_name = "${azurerm_virtual_network.myterraformnetwork.name}"
    address_prefix       = "${var.aks_subnet_address_prefix}"
    network_security_group_id = "${azurerm_network_security_group.aks_nsg.id}"
}

# Create Azure Kubernetes Service Cluster
resource "azurerm_log_analytics_workspace" "myterraform_log_analytics_workspace" {
    # The WorkSpace name has to be unique across the whole of azure, not just the current subscription/tenant.
    name                = "${var.log_analytics_workspace_name}"
    location            = "${azurerm_resource_group.myterraformgroup.location}"
    resource_group_name = "${azurerm_resource_group.myterraformgroup.name}"
    sku                 = "${var.log_analytics_workspace_sku}"
}
resource "azurerm_log_analytics_solution" "myterraform_log_analytics_solution" {
    solution_name         = "ContainerInsights"
    location              = "${azurerm_resource_group.myterraformgroup.location}"
    resource_group_name   = "${azurerm_resource_group.myterraformgroup.name}"
    workspace_resource_id = "${azurerm_log_analytics_workspace.myterraform_log_analytics_workspace.id}"
    workspace_name        = "${azurerm_log_analytics_workspace.myterraform_log_analytics_workspace.name}"

    plan {
        publisher = "Microsoft"
        product   = "OMSGallery/ContainerInsights"
    }
}
resource "azurerm_kubernetes_cluster" "myterraform_kubernetes_cluster" {
    name                = "${var.azure_kubernetes_service_cluster_name}"
    location            = "${azurerm_resource_group.myterraformgroup.location}"
    resource_group_name = "${azurerm_resource_group.myterraformgroup.name}"
    dns_prefix          = "${var.azure_kubernetes_service_dns_prefix}"

    kubernetes_version = "${var.azure_kubernetes_service_cluster_version}"

    linux_profile {
        admin_username = "${var.admin_username}"

        ssh_key {
            key_data = "${var.ssh_public_key_data}"
        }
    }

    agent_pool_profile {
        name            = "agentpool"
        count           = "${var.agent_count}"
        vm_size         = "Standard_D4s_v3"
        os_type         = "Linux"
        os_disk_size_gb = 30
        vnet_subnet_id  = "${azurerm_subnet.aks_subnet.id}"
    }

    network_profile {
        network_plugin = "azure"
        network_policy = "azure"
        dns_service_ip = "10.0.0.10"
        docker_bridge_cidr = "172.17.0.1/16"
        service_cidr = "10.0.0.0/16"
        load_balancer_sku = "standard"
    }

    role_based_access_control {
        enabled = "true"
    }

    addon_profile {
        oms_agent {
            enabled                    = "true"
            log_analytics_workspace_id = "${azurerm_log_analytics_workspace.myterraform_log_analytics_workspace.id}"
        }
    }
    
    service_principal {
        client_id       = "${var.client_id}"
        client_secret   = "${var.client_secret}"
    }

    tags = {
        Environment = "Azure Terraform Kubernetes Service"
    }
}

# Create Azure Container Registry
resource "azurerm_container_registry" "myterraform_container_registry" {
  name                     = "${var.azure_container_registry_name}"
  resource_group_name      = "${azurerm_resource_group.myterraformgroup.name}"
  location                 = "${azurerm_resource_group.myterraformgroup.location}"
  sku                      = "Premium"
  admin_enabled            = false
}

# Create AKS JumperVM
resource "azurerm_public_ip" "aks_jumpervm0001_pip" {
    name                         = "${var.aks_jumpervm_name}"
    location                     = "${azurerm_resource_group.myterraformgroup.location}"
    resource_group_name          = "${azurerm_resource_group.myterraformgroup.name}"
    allocation_method            = "Static"

    tags = {
        environment = "Azure Terraform Kubernetes Service"
    }
}
resource "azurerm_network_interface" "aks_jumpervm0001_nic" {
    name                        = "${var.aks_jumpervm_nic_name}"
    location                    = "${azurerm_resource_group.myterraformgroup.location}"
    resource_group_name         = "${azurerm_resource_group.myterraformgroup.name}"

    ip_configuration {
        name                          = "${var.aks_jumpervm_nic_name}"
        subnet_id                     = "${azurerm_subnet.aks_subnet.id}"
        private_ip_address_allocation = "Dynamic"
        public_ip_address_id          = "${azurerm_public_ip.aks_jumpervm0001_pip.id}"
    }

    tags = {
        environment = "Azure Terraform Kubernetes Service"
    }
}

# Create virtual machine
resource "azurerm_virtual_machine" "aksjumpervm0001" {
    name                  = "${var.aks_jumpervm_name}"
    location              = "${azurerm_resource_group.myterraformgroup.location}"
    resource_group_name   = "${azurerm_resource_group.myterraformgroup.name}"
    network_interface_ids = ["${azurerm_network_interface.aks_jumpervm0001_nic.id}"]
    vm_size               = "Standard_D2s_v3"

    storage_os_disk {
        name              = "${var.aks_jumpervm_name}OsDisk"
        caching           = "ReadWrite"
        create_option     = "FromImage"
        managed_disk_type = "Standard_LRS"
    }

    storage_image_reference {
        id = "/subscriptions/xxxx-xxxx-xxxx-xxxx-xxxx/resourceGroups/xxxx-xxxx/providers/Microsoft.Compute/images/xxxx"
    }

    os_profile {
        computer_name  = "${var.aks_jumpervm_name}"
        admin_username = "azureuser"
    }

    os_profile_linux_config {
        disable_password_authentication = true
        ssh_keys {
            path     = "/home/azureuser/.ssh/authorized_keys"
            key_data = "${var.ssh_public_key_data}"
        }
    }
}

所有资源会在 10 分钟以内部署完成。

3.2 获取 AKS Credential 并查看 AKS Cluster 信息：

1	az aks get-credentials --resource-group aksrg001 --name akscluster001

集群节点信息：

$ kubectl get node
NAME                       STATUS   ROLES   AGE   VERSION
aks-agentpool-26667448-0   Ready    agent   17m   v1.14.8
aks-agentpool-26667448-1   Ready    agent   16m   v1.14.8
aks-agentpool-26667448-2   Ready    agent   16m   v1.14.8

1	docker login aksacr001.azurecr.io

3.4 Flink Streaming on AKS

3.4.1 Streaming 任务示例

该任务会从某个端口中读取文本，分割为单词，并且每 5 秒钟打印一次每个单词出现的次数。以下代码是从 Flink 官方文档上获取来的。

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

public class WindowWordCount {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStream<Tuple2<String, Integer>> dataStream = env
                .socketTextStream("10.11.4.4", 9999)
                .flatMap(new Splitter())
                .keyBy(0)
                .timeWindow(Time.seconds(5))
                .sum(1);

        dataStream.print();

        env.execute("Window WordCount");
    }

    public static class Splitter implements FlatMapFunction<String, Tuple2<String, Integer>> {
        @Override
        public void flatMap(String sentence, Collector<Tuple2<String, Integer>> out) throws Exception {
            for (String word: sentence.split(" ")) {
                out.collect(new Tuple2<String, Integer>(word, 1));
            }
        }
    }

}

接下来执行 mvn clean package 命令，打包好的 Jar 文件路径为 target/flink-on-kubernetes-1.0.0-SNAPSHOT-jar-with-dependencies.jar。

3.4.2 构建自定义 Flink Docker 镜像并推送至 ACR

Flink 在 DockerHub 上提供了一个官方的容器镜像，我们可以以该镜像为基础，构建独立的 Flink 镜像，主要就是将自定义 Jar 包打到镜像里面去。此外，新版 Flink 已将 Hadoop 依赖从官方发行版中剥离，因此我们在打镜像时也需要包含进去。官方的 Dockerfile 主要做了将 OpenJDK 1.8 作为 JDK 环境，下载并安装 Flink 至 /opt/flink，同时添加 flink 用户和组等。我们在此基础上构建自定义 Flink 镜像：

FROM flink:1.9.1-scala_2.12

ARG hadoop_jar
ARG job_jar

ENV FLINK_CONF=$FLINK_HOME/conf/flink-conf.yaml

RUN set -x && \
  sed -i -e "s/jobmanager\.heap\.size:.*/jobmanager.heap.size: 128m/g" $FLINK_CONF && \
  sed -i -e "s/taskmanager\.heap\.size:.*/taskmanager.heap.size: 256m/g" $FLINK_CONF

COPY --chown=flink:flink $hadoop_jar $job_jar $FLINK_HOME/lib/

USER flink

注：hadoop_jar 从这里下载；job_jar 为刚刚编译打包好的 flink-on-kubernetes-1.0.0-SNAPSHOT-jar-with-dependencies.jar。下载好后做编译：

$ cd /path/to/Dockerfile
$ cp /path/to/flink-shaded-hadoop-2-uber-2.8.3-7.0.jar hadoop.jar
$ cp /path/to/flink-on-kubernetes-1.0.0-SNAPSHOT-jar-with-dependencies.jar job.jar
$ docker build --build-arg hadoop_jar=hadoop.jar --build-arg job_jar=job.jar --tag flink-on-kubernetes:1.0.0 .

查看镜像：

1
2
3

$ docker image ls
REPOSITORY            TAG                 IMAGE ID            CREATED              SIZE
flink-on-kubernetes   1.0.0               25a02549b943        8 seconds ago        575MB

镜像推送至 ACR：

1 2	docker tag flink-on-kubernetes:1.0.0 aksacr001.azurecr.io/flink-on-kubernetes:v1 docker push aksacr001.azurecr.io/flink-on-kubernetes:v1

3.4.3 构建 Flink Streaming JobManager 和 Service

$ vi flink-streaming-on-aks-jobmanager.yml
apiVersion: batch/v1
kind: Job
metadata:
  name: ${JOB}-jobmanager
spec:
  template:
    metadata:
      labels:
        app: flink
        instance: ${JOB}-jobmanager
    spec:
      restartPolicy: OnFailure
      containers:
      - name: jobmanager
        image: aksacr001.azurecr.io/flink-on-kubernetes:v1
        command: ["/opt/flink/bin/standalone-job.sh"]
        args: ["start-foreground",
               "-Djobmanager.rpc.address=${JOB}-jobmanager",
               "-Dparallelism.default=1",
               "-Dblob.server.port=6124",
               "-Dqueryable-state.server.ports=6125"]
        ports:
        - containerPort: 6123
          name: rpc
        - containerPort: 6124
          name: blob
        - containerPort: 6125
          name: query
        - containerPort: 8081
          name: ui

使用 kubectl 命令创建对象，并查看状态：

$ export JOB=flink-streaming-on-aks
$ envsubst <flink-streaming-on-aks-jobmanager.yml | kubectl apply -f -
$ kubectl get pods
NAME                                      READY   STATUS    RESTARTS   AGE
flink-streaming-on-aks-jobmanager-xhkkc   1/1     Running   0          48s

接着创建一个 Service 来将 JobManager 的端口开放出来，以便 TaskManager 做服务注册和调用：

$ vi flink-streaming-on-aks-service.yml
apiVersion: v1
kind: Service
metadata:
  name: ${JOB}-jobmanager
spec:
  selector:
    app: flink
    instance: ${JOB}-jobmanager
  type: NodePort
  ports:
  - name: rpc
    port: 6123
  - name: blob
    port: 6124
  - name: query
    port: 6125
  - name: ui
    port: 8081

$ envsubst <flink-streaming-on-aks-service.yml | kubectl apply -f -
$ kubectl get svc flink-streaming-on-aks-jobmanager
NAME                                TYPE       CLUSTER-IP   EXTERNAL-IP   PORT(S)                                                       AGE
flink-streaming-on-aks-jobmanager   NodePort   10.0.25.24   <none>        6123:32291/TCP,6124:30580/TCP,6125:30458/TCP,8081:30855/TCP   46s

3.4.4 构建 Flink Streaming TaskManager

$ vi flink-streaming-on-aks-taskmanager.yml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: ${JOB}-taskmanager
spec:
  selector:
    matchLabels:
      app: flink
      instance: ${JOB}-taskmanager
  replicas: 3
  template:
    metadata:
      labels:
        app: flink
        instance: ${JOB}-taskmanager
    spec:
      containers:
      - name: taskmanager
        image: aksacr001.azurecr.io/flink-on-kubernetes:v1
        command: ["/opt/flink/bin/taskmanager.sh"]
        args: ["start-foreground", "-Djobmanager.rpc.address=${JOB}-jobmanager"]

$ envsubst <flink-streaming-on-aks-taskmanager.yml | kubectl apply -f -
$ kubectl get pods
NAME                                                 READY   STATUS              RESTARTS   AGE
flink-streaming-on-aks-jobmanager-xhkkc              1/1     Running             0          19m
flink-streaming-on-aks-taskmanager-b9df9657d-2vnrt   0/1     Running   0          5s
flink-streaming-on-aks-taskmanager-b9df9657d-5nx95   0/1     Running   0          6s
flink-streaming-on-aks-taskmanager-b9df9657d-g2p7w   0/1     Running   0          5s

目前启动了 JobManager 和 TaskManager 的 AKS Pods 作为整个 Streaming 流计算任务的计算资源，下面进行实时计算任务测试。Flink WebUI如下：

Flink JobManager 的 WebUI 可以清晰看到 Job 以及整个 Flink Job 的 Pipeline，包括所有的 Log 和 GC 的情况也有显示。

3.4.5 Flink Streaming Job 运行和结果

还记得 3.6.1 的 Java 代码吗？此时我们需要通过 nc 命令打开 9999 端口，然后输入 messages (输出系统日志来模拟) ：

$ nc -lk 9999
Sep 18 05:17:16 localhost journal: Runtime journal is using 8.0M (max allowed 398.2M, trying to leave 597.3M free of 3.8G available → current limit 398.2M).

Sep 18 05:17:16 localhost kernel: Initializing cgroup subsys cpuset
Sep 18 05:17:16 localhost kernel: Initializing cgroup subsys cpu
Sep 18 05:17:16 localhost kernel: Initializing cgroup subsys cpuacct

Sep 18 05:17:16 localhost kernel: BIOS-e820: [mem 0x0000000000000000-0x000000000009fbff] usable
Sep 18 05:17:16 localhost kernel: BIOS-e820: [mem 0x000000000009fc00-0x000000000009ffff] reserved
Sep 18 05:17:16 localhost kernel: BIOS-e820: [mem 0x00000000000e0000-0x00000000000fffff] reserved
Sep 18 05:17:16 localhost kernel: BIOS-e820: [mem 0x0000000000100000-0x000000003ffeffff] usable

Sep 18 05:17:16 localhost kernel: DMI: Microsoft Corporation Virtual Machine/Virtual Machine, BIOS 090007  06/02/2017
Sep 18 05:17:16 localhost kernel: Hypervisor detected: Microsoft HyperV
Sep 18 05:17:16 localhost kernel: HyperV: features 0x2e7f, hints 0x44c2c
Sep 18 05:17:16 localhost kernel: Hyper-V Host Build:14393-10.0-0-0.299
Sep 18 05:17:16 localhost kernel: HyperV: LAPIC Timer Frequency: 0x30d40

Flink Streaming Job 实时结果如下：

$ kubectl logs -f -l instance=$JOB-taskmanager
(,4)
(Sep,1)
(398.2M).,1)
(limit,1)
(current,1)
(→,1)
(available,1)
(3.8G,1)
(of,1)
(free,1)
(597.3M,1)
(leave,1)
(to,1)
(trying,1)
(398.2M,,1)
(allowed,1)
((max,1)
(8.0M,1)
(using,1)
(is,1)
(journal,1)
(Runtime,1)
(journal:,1)
(localhost,1)
(05:17:16,1)
(18,1)
(,2)
(cpuacct,1)
(cpu,1)
(cpuset,1)
(subsys,3)
(cgroup,3)
(Initializing,3)
(kernel:,3)
(localhost,3)
(05:17:16,3)
(18,3)
(Sep,3)
(Sep,4)
(0x0000000000100000-0x000000003ffeffff],1)
(0x00000000000e0000-0x00000000000fffff],1)
(reserved,2)
(0x000000000009fc00-0x000000000009ffff],1)
(usable,2)
(0x0000000000000000-0x000000000009fbff],1)
([mem,4)
(BIOS-e820:,4)
(kernel:,4)
(localhost,4)
(05:17:16,4)
(18,4)
(,1)
(Sep,5)
(0x30d40,1)
(Frequency:,1)
(Timer,1)
(LAPIC,1)
(Build:14393-10.0-0-0.299,1)
(Host,1)
(Hyper-V,1)
(0x44c2c,1)
(hints,1)
(0x2e7f,,1)
(features,1)
(HyperV:,2)
(HyperV,1)
(detected:,1)
(Hypervisor,1)
... 后面结果省略了 ...

随着实时任务不断的输入追加，整个 Flink Streaming Job 都处于 Running 状态，永远都不会停下来。

3.5 Flink Batch on AKS

玩完 Streaming 流计算，我们再来看看批处理任务。其实批处理任务和流计算任务本属一家，只是时间维度上的特例，所以这也是 Flink 对流批任务处理的思想的统一。下面我们也运行同样的 WordCount 批处理任务来玩玩 Flink Batch on AKS。

3.5.1 下载 Flink 二进制文件

从 Flink 官网选择合适的版本下载，本例中使用 1.9.1 版本。

1 2	wget https://www-us.apache.org/dist/flink/flink-1.9.1/flink-1.9.1-bin-scala_2.12.tgz tar -zvxf flink-1.9.1-bin-scala_2.12.tgz

3.5.1 编写 ConfigMap Yaml 文件

$ vi flink-batch-on-aks-configmap.yaml
apiVersion: v1
kind: ConfigMap
metadata:
  name: flink-batch-on-aks-configmap
  labels:
    app: flink
data:
  flink-conf.yaml: |+
    jobmanager.rpc.address: flink-jobmanager
    taskmanager.numberOfTaskSlots: 1
    blob.server.port: 6124
    jobmanager.rpc.port: 6123
    taskmanager.rpc.port: 6122
    jobmanager.heap.size: 1024m
    taskmanager.heap.size: 1024m
  log4j.properties: |+
    log4j.rootLogger=INFO, file
    log4j.logger.akka=INFO
    log4j.logger.org.apache.kafka=INFO
    log4j.logger.org.apache.hadoop=INFO
    log4j.logger.org.apache.zookeeper=INFO
    log4j.appender.file=org.apache.log4j.FileAppender
    log4j.appender.file.file=${log.file}
    log4j.appender.file.layout=org.apache.log4j.PatternLayout
    log4j.appender.file.layout.ConversionPattern=%d{yyyy-MM-dd HH:mm:ss,SSS} %-5p %-60c %x - %m%n
    log4j.logger.org.apache.flink.shaded.akka.org.jboss.netty.channel.DefaultChannelPipeline=ERROR, file

创建并查看 ConfigMap：

$ kubectl apply -f flink-batch-on-aks-configmap.yaml
$ kubectl get cm
NAME                           DATA   AGE
flink-batch-on-aks-configmap   2      25s

3.5.2 编写 JobManager Yaml 文件

$ vi flink-batch-on-aks-jobmanager.yaml 
apiVersion: extensions/v1beta1
kind: Deployment
metadata:
  name: flink-batch-on-aks-jobmanager
spec:
  replicas: 1
  template:
    metadata:
      labels:
        app: flink
        component: jobmanager
    spec:
      containers:
      - name: jobmanager
        image: aksacr001.azurecr.io/flink-on-kubernetes:v1
        workingDir: /opt/flink
        command: ["/bin/bash", "-c", "$FLINK_HOME/bin/jobmanager.sh start;\
          while :;
          do
            if [[ -f $(find log -name '*jobmanager*.log' -print -quit) ]];
              then tail -f -n +1 log/*jobmanager*.log;
            fi;
          done"]
        ports:
        - containerPort: 6123
          name: rpc
        - containerPort: 6124
          name: blob
        - containerPort: 8081
          name: ui
        livenessProbe:
          tcpSocket:
            port: 6123
          initialDelaySeconds: 30
          periodSeconds: 60
        volumeMounts:
        - name: flink-config-volume
          mountPath: /opt/flink/conf
      volumes:
      - name: flink-config-volume
        configMap:
          name: flink-batch-on-aks-configmap
          items:
          - key: flink-conf.yaml
            path: flink-conf.yaml
          - key: log4j.properties
            path: log4j.properties

创建并查看 JobManager :

$ kubectl apply -f flink-batch-on-aks-jobmanager.yaml
$ kubectl get pods
NAME                                             READY   STATUS    RESTARTS   AGE
flink-batch-on-aks-jobmanager-7ddb5cf4bc-nh7b8   1/1     Running   0          7s

3.5.3 编写 TaskManager Yaml 文件

$ vi flink-batch-on-aks-taskmanager.yml
apiVersion: extensions/v1beta1
kind: Deployment
metadata:
  name: flink-batch-on-aks-taskmanager
spec:
  replicas: 2
  template:
    metadata:
      labels:
        app: flink
        component: taskmanager
    spec:
      containers:
      - name: taskmanager
        image: aksacr001.azurecr.io/flink-on-kubernetes:v1
        workingDir: /opt/flink
        command: ["/bin/bash", "-c", "$FLINK_HOME/bin/taskmanager.sh start; \
          while :;
          do
            if [[ -f $(find log -name '*taskmanager*.log' -print -quit) ]];
              then tail -f -n +1 log/*taskmanager*.log;
            fi;
          done"]
        ports:
        - containerPort: 6122
          name: rpc
        livenessProbe:
          tcpSocket:
            port: 6122
          initialDelaySeconds: 30
          periodSeconds: 60
        volumeMounts:
        - name: flink-config-volume
          mountPath: /opt/flink/conf/
      volumes:
      - name: flink-config-volume
        configMap:
          name: flink-batch-on-aks-configmap
          items:
          - key: flink-conf.yaml
            path: flink-conf.yaml
          - key: log4j.properties
            path: log4j.properties

创建并查看 TaskManager：

$ kubectl apply -f flink-batch-on-aks-taskmanager.yaml
$ kubectl get pods
NAME                                             READY   STATUS    RESTARTS   AGE
flink-batch-on-aks-jobmanager-7ddb5cf4bc-nh7b8   1/1     Running   0          2m28s
flink-batch-on-aks-taskmanager-cdc4498b9-6pctv   1/1     Running   0          11s
flink-batch-on-aks-taskmanager-cdc4498b9-kbkdf   1/1     Running   0          11s

3.5.4 编写 JobManager Service Yaml 文件

$ vi flink-batch-on-aks-jobmanager-service.yaml 
apiVersion: v1
kind: Service
metadata:
  name: flink-batch-on-aks-jobmanager
spec:
  type: ClusterIP
  ports:
  - name: rpc
    port: 6123
  - name: blob
    port: 6124
  - name: ui
    port: 8081
  selector:
    app: flink
    component: jobmanager

创建并查看 TaskManager Service :

$ kubectl apply -f flink-batch-on-aks-jobmanager-service.yaml 
$ kubectl get svc
NAME                            TYPE        CLUSTER-IP    EXTERNAL-IP   PORT(S)                      AGE
flink-batch-on-aks-jobmanager   ClusterIP   10.0.23.214   <none>        6123/TCP,6124/TCP,8081/TCP   7s

3.5.5 启动 Flink UI ：

1	$ kubectl port-forward flink-batch-on-aks-jobmanager-7ddb5cf4bc-nh7b8 --address 0.0.0.0 8081:8081

3.5.6 提交 Flink Batch Job 到 AKS 集群：

$ ./bin/flink run -m 10.11.4.4:8081 ./examples/streaming/WordCount.jar --input ./input.txt
Starting execution of program
Printing result to stdout. Use --output to specify output path.
Program execution finished
Job with JobID 7a9037a6c6573a6a7f01f36e6e9d7382 has finished.
Job Runtime: 138 ms

可以通过 Flink Batch UI 查看计算结果。

4. 总结

Flink 的思想是把一切任务都看作流来实现流批的计算。本文举例了 2 个场景，验证了在 AKS 上如何运行 Flink Streaming & Batch 作业，希望能够给大家一些参考。但是本文也有很多其他场景并未考虑到，比如：JobManager HA、External Checkpoints (可以和 Azure Blob Storage集成)、动态的任务扩容、与 Eventhub 或 Kafka 的集成等等，这些场景留待大家自己发现和实践了。

2019-10-13

All Blogs►Microsoft Azure►Kubernetes►Bigdata►Spark►Linux

强强联合! 利用 Microsoft Azure AKS 集群集成 Apache Spark 来做大数据计算

1. Azure Kubernetes Service 及 Apache Spark 简介

伴随着 Kubernetes 技术的日渐成熟和社区生态的蓬勃发展, 越来越多的客户都将自己很多的应用程序迁移到容器服务上, 以便能够设计出更加快速、便捷、更高可靠的应用架构. 为了提高 Kubernetes 集群的部署速度以及降低其运维成本, Microsoft Azure 推出了托管的 Kubernetes 服务 —— Azure Kubernetes Service (AKS). AKS 带来了诸多优点, 如高可用的控制平面、Calico 等网络插件集成、Azure 服务无缝对接、与开源版本 Kubernetes 同步匹配、平滑迁移及无感知升级等。

在大数据处理与分析领域, 从技术的成熟度、稳定度以及社区活度程度来看, Apache Spark 是目前最流行的计算框架并在实际生产中处于重要地位并广泛使用. Apache Spark 支持 Standalone、Mesos 以及 YARN 等集群资源管理调度平台, 其中以 YARN 在实际生产中使用最为广泛, YARN 就是我们熟知的 Hadoop 平台的资源调度框架. 在计算存储不分离的架构下, 集群每个节点也会作为 Datanode 即采用 HDFS 分布式文件系统来进行数据存储. 但是随着数据量的爆炸式增长以及数据湖概念的日益普及, 计算与存储分离也逐渐成为了刚需. 为此, Microsoft Azure 也推出了满足以上需求的托管式 Hadoop 服务 —— Azure HDInsight 以及 Databricks. HDInsight 是 HDP 版本 Hadoop 在 Azure 云上的托管服务, Databricks 是著名 Spark 创始人完全基于 Apache Spark 并针对 Microsoft Azure 云服务平台进行优化的 Spark 托管服务.

目前最火热的两个技术已经可以结合并迸发出了新的火花, 2.3.0 版本以上的 Apache Spark 已经提供了对 Kubernetes 平台的支持与集成, 使我们能够利用原生的 Kubernetes 来进行集群计算资源的分配与管理 Spark 计算作业. 在该模式下, Spark Driver 和 Executor 都通过 Kubernetes Pods 来运行, 并且也可通过指定 Node Selector 将计算作业运行于特定节点之上 (例如: 带有GPU的实例类型等).

（from: https://spark.apache.org).

2. Apache Spark on AKS 实现

本实验需要正确安装 azcli, kubectl 等命令行工具, 详细步骤请参考相应文档, 本文不再赘述. 具体的操作过程在 AKS Vnet 的一台内网服务器的 root 家目录下进行, 该服务器作为 Spark 集群的 Gateway 客户端.

2.1 创建 AKS 集群

2.1.1 创建资源组

1	az group create --name akssparkrg --location southeastasia

2.1.2 创建 3 节点 AKS 集群

az aks create --resource-group akssparkrg \
--name aksspark0001 \
--kubernetes-version 1.14.7 \
--node-count 3 \
--node-vm-size Standard_DS2_v3 \
--enable-addons monitoring \
--admin-username akssparkuser \
--ssh-key-value ~/.ssh/id_rsa.pub

集群创建时间在20分钟左右.

2.1.3 获取 AKS Credential:

1	az aks get-credentials --resource-group akssparkrg --name aksspark0001

2.2 在 AKS 集群中创建 Spark 服务帐户并绑定权限

2.2.1 创建 Spark 服务账号

1	kubectl create serviceaccount spark

2.2.2 权限绑定

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

2.3 下载 Spark 二进制文件

从 Spark官网选择合适的版本下载, 本例中使用 2.4.4 版本 ( 需要高于2.3.0 ).

1 2	wget https://www-eu.apache.org/dist/spark/spark-2.4.4/spark-2.4.4-bin-hadoop2.7.tgz tar -zvxf spark-2.4.4-bin-hadoop2.7.tgz

2.4 确认 Spark Jar 和 Kubernetes 的版本兼容性

Spark 采用了 fabric8.io 的 Kubernetes & OpenShift Java Client 客户端连接 API Server 来进行资源调度和分配, 需要先确认 Spark 2.4.4 自带的 kubernetes-client-*.jar 版本与 AKS 集群版本是否兼容. Spark 2.4.4 自带的 Kubernetes-client 版本为 4.1.2 ( kubernetes-client-4.1.2.jar ), AKS集群版本为 1.14.7, 存在版本不兼容问题，需要更高版本 jar 包. 具体的版本对应关系如下表所示:

|                           | Kubernetes 1.4.9 | Kubernetes 1.6.0 | Kubernetes 1.7.0  | Kubernetes 1.9.0  | Kubernetes 1.10.0 | Kubernetes 1.11.0 | Kubernetes 1.12.0 | Kubernetes 1.14.2 | Kubernetes 1.15.3 |
|---------------------------|------------------|------------------|-------------------|-------------------|-------------------|-------------------|-------------------|-------------------|-------------------|
| kubernetes-client 1.3.92  | +                | +                | -                 | -                 | -                 | -                 | -                 | -                 | -                 |
| kubernetes-client 3.0.3   | -                | -                | ✓                 | -                 | -                 | -                 | -                 | -                 | -                 |
| kubernetes-client 3.0.10  | -                | ✓                | ✓                 | ✓                 | -                 | -                 | -                 | -                 | -                 |
| kubernetes-client 3.0.11  | -                | ✓                | ✓                 | ✓                 | -                 | -                 | -                 | -                 | -                 |
| kubernetes-client 3.1.12  | -                | ✓                | ✓                 | ✓                 | -                 | -                 | -                 | -                 | -                 |
| kubernetes-client 3.2.0   | -                | ✓                | ✓                 | ✓                 | -                 | -                 | -                 | -                 | -                 |
| kubernetes-client 4.0.0   | -                | ✓                | ✓                 | ✓                 | -                 | -                 | -                 | -                 | -                 |
| kubernetes-client 4.1.0   | -                | ✓                | ✓                 | ✓                 | -                 | -                 | -                 | -                 | -                 |
| kubernetes-client 4.1.1   | -                | -                | -                 | ✓                 | ✓                 | ✓                 | ✓                 | -                 | -                 |
| kubernetes-client 4.1.2   | -                | -                | -                 | ✓                 | ✓                 | ✓                 | ✓                 | -                 | -                 |
| kubernetes-client 4.1.3   | -                | -                | -                 | ✓                 | ✓                 | ✓                 | ✓                 | -                 | -                 |
| kubernetes-client 4.2.0   | -                | -                | -                 | ✓                 | ✓                 | ✓                 | ✓                 | -                 | -                 |
| kubernetes-client 4.2.1   | -                | -                | -                 | ✓                 | ✓                 | ✓                 | ✓                 | -                 | -                 |
| kubernetes-client 4.2.2   | -                | -                | -                 | ✓                 | ✓                 | ✓                 | ✓                 | -                 | -                 |
| kubernetes-client 4.3.0   | -                | -                | -                 | ✓                 | ✓                 | ✓                 | ✓                 | ✓                 | -                 |
| kubernetes-client 4.3.1   | -                | -                | -                 | ✓                 | ✓                 | ✓                 | ✓                 | ✓                 | -                 |
| kubernetes-client 4.4.0   | -                | -                | -                 | ✓                 | ✓                 | ✓                 | ✓                 | ✓                 | -                 |
| kubernetes-client 4.4.1   | -                | -                | -                 | ✓                 | ✓                 | ✓                 | ✓                 | ✓                 | -                 |
| kubernetes-client 4.4.2   | -                | -                | -                 | ✓                 | ✓                 | ✓                 | ✓                 | ✓                 | -                 |
| kubernetes-client 4.5.0   | -                | -                | -                 | ✓                 | ✓                 | ✓                 | ✓                 | ✓                 | -                 |
| kubernetes-client 4.5.1   | -                | -                | -                 | ✓                 | ✓                 | ✓                 | ✓                 | ✓                 | -                 |
| kubernetes-client 4.5.2   | -                | -                | -                 | ✓                 | ✓                 | ✓                 | ✓                 | ✓                 | -                 |
| kubernetes-client 4.6.0   | -                | -                | -                 | ✓                 | ✓                 | ✓                 | ✓                 | ✓                 | ✓                 |

本文我们直接使用能够与 Kubernetes 1.14.7 版本兼容的 kubernetes-client 4.6.0.jar, 通过这里下载, 下载后替换 spark-2.4.4-bin-hadoop2.7/jars 目录下的 kubernetes-client-4.1.2.jar 即可.

2.5 创建 ACR (Azure Container Registry) 并与 AKS 集成

2.5.1 创建 ACR

1	az acr create -n akssparkacr0001 -g akssparkrg --sku basic

记录 ACR Resource ID:
/subscriptions/53a326cc-f961-4540-8701-2bfd1003242b/resourceGroups/akssparkrg/providers/Microsoft.ContainerRegistry/registries/akssparkacr0001

2.5.2 ACR 与 AKS 集群集成

1
2

az aks update -n aksspark0001 -g akssparkrg --attach-acr akssparkacr0001
az aks update -n aksspark0001 -g akssparkrg --attach-acr /subscriptions/53a326cc-f961-4540-8701-2bfd1003242b/resourceGroups/akssparkrg/providers/Microsoft.ContainerRegistry/registries/akssparkacr0001

2.5.3 登录 ACR

1	az acr login -n akssparkacr0001

2.6 构建 Spark 镜像

2.6.1 构建并推送镜像至 ACR

REGISTRY_NAME=akssparkacr0001.azurecr.io
REGISTRY_TAG=v1

# 执行脚本构建 Spark 镜像
./bin/docker-image-tool.sh -r $REGISTRY_NAME -t $REGISTRY_TAG build

# 推送镜像
./bin/docker-image-tool.sh -r $REGISTRY_NAME -t $REGISTRY_TAG push

2.6.2 Azure 查看docker image:

通过 docker image list 也可以看, 自动构建了 Spark 原生, Spark-R 以及 Pyspark 的容器镜像.

2.7 运行作业 SparkPi

Spark 作业两种方式来运行, Cluster 和 Client 模式. 两种模式的主要区别简单概括就是 Spark Job Driver 是在本地节点还是集群节点上. 在实际生产过程中, 通常会将 spark-submit 的作业提交方式服务化, 通过外部服务调用将作业提交到集群上运行, 最典型的就是 Apache Livy. 本文采取通过命令行 spark-submit 将 SparkPi 作业提交到集群上计算的方式.

2.7.1 创建 Storage Account 并将 spark-examples_2.11-2.4.4.jar 上传至 Azure Blob

# 变量定义, storage account name
RESOURCE_GROUP=akssparkrg
STORAGE_ACCT=akssparksa0001

# 创建 storage account
az group create --name $RESOURCE_GROUP --location southeastasia
az storage account create --resource-group $RESOURCE_GROUP --name $STORAGE_ACCT --sku Standard_LRS
export AZURE_STORAGE_CONNECTION_STRING=`az storage account show-connection-string --resource-group $RESOURCE_GROUP --name $STORAGE_ACCT -o tsv`

# 指定上传 jar
CONTAINER_NAME=akssparkjars
BLOB_NAME=spark-examples_2.11-2.4.4.jar
FILE_TO_UPLOAD=/root/aksspark/spark-2.4.4-bin-hadoop2.7/examples/jars/spark-examples_2.11-2.4.4.jar

# 创建 blob container
echo "Creating the container..."
az storage container create --name $CONTAINER_NAME
az storage container set-permission --name $CONTAINER_NAME --public-access blob

# 上传 jar
echo "Uploading the file..."
az storage blob upload --container-name $CONTAINER_NAME --file $FILE_TO_UPLOAD --name $BLOB_NAME

# 获取 jar url 
JARURL=$(az storage blob url --container-name $CONTAINER_NAME --name $BLOB_NAME | tr -d '"')

2.7.2 运行 SparkPi

通过 spark-submit 提交 SparkPi 作业

cd spark-2.4.4-bin-hadoop2.7
./bin/spark-submit \
  --master k8s://https://aksspark00-akssparkrg-53a326-07f35845.hcp.southeastasia.azmk8s.io:443 \ # AKS 集群地址
  --deploy-mode cluster \ # 集群模式
  --name spark-pi \
  --class org.apache.spark.examples.SparkPi \
  --conf spark.executor.instances=3 \ # 申请 3 个 Executor
  --conf spark.kubernetes.container.image=akssparkacr0001.azurecr.io/spark:v1 \ # Docker Image 地址
  --conf spark.kubernetes.authenticate.driver.serviceAccountName=spark \ # 指定创建的 Service Account spark
  https://akssparksa0001.blob.core.windows.net/akssparkjars/spark-examples_2.11-2.4.4.jar # Jar

spark-submit 的具体参数可以参考 Spark on Kubernetes官网, 不过官方文档只有默认参数配置. 如果需要看所有参数的可选值, 可以考虑去看 GitHub 上 Spark 源代码.

2.7.3 查看运行结果

查看 Spark Job Driver 和对应的 Executor Pods

[root@AKSSpark spark-2.4.4-bin-hadoop2.7]# kubectl get pods
NAME                            READY   STATUS    RESTARTS   AGE
spark-pi-1571055145415-driver   1/1     Running   0          25s
spark-pi-1571055145415-exec-1   1/1     Running   0          15s
spark-pi-1571055145415-exec-2   1/1     Running   0          15s
spark-pi-1571055145415-exec-3   0/1     Pending   0          15s

运行作业时. 还可以访问 Spark UI. kubectl port-forward 命令提供对 Spark Job UI 的访问权限.

1	kubectl port-forward spark-pi-1571055145415-driver 4040:4040

获取作业结果和日志

1
2
3

[root@AKSSpark spark-2.4.4-bin-hadoop2.7]# kubectl get pods
NAME                                       READY   STATUS      RESTARTS   AGE
spark-pi-1571055145415-driver              0/1     Completed   0          53s

使用 kubectl logs 来获取 Spark 作业的 Log

1	kubectl logs spark-pi-1571055145415-driver

日志中，可以看到 Spark 作业的结果，即 Pi 的值.

1	Pi is roughly 3.1412957064785325

2.8 使用 Spark Shell 进行交互式分析 Azure Blob 数据

在很多需要调试的场景中, 都会通过命令行的交互式界面来进行调试. 对于 Spark 来说, 有 Spark Shell、Spark-R、Pyspark 三种命令行, 分别针对 Scala, R 以及 Python. 下面我们通过Spark Shell 来分析 Azure Blob 的数据.

2.8.1 Docker Image 新打包 v2 版本

Spark 通过 wasbs://{container-name}@{storage-account-name}.blob.core.windows.net 访问 Azure Blob, 这需要在 Spark Jar 中加载 azure-storage.jar 和 hadoop-azure.jar, 这些 Jar 需要打在 Docker Image 里面, 然后上传至 ACR.

1
2
3

cd spark-2.4.4-bin-hadoop2.7/jars
wget http://central.maven.org/maven2/com/microsoft/azure/azure-storage/2.2.0/azure-storage-2.2.0.jar
wget http://central.maven.org/maven2/org/apache/hadoop/hadoop-azure/2.7.3/hadoop-azure-2.7.3.jar

然后重复 2.6 节的操作, 指定版本号为 v2, 打包好后推送到 ACR 上, 具体的 Image 地址为 akssparkacr0001.azurecr.io/spark:v2.

2.8.2 使用 Spark Shell 交互式分析 Azure Blob 中的数据

运行 Spark Shell

cd spark-2.4.4-bin-hadoop2.7
./bin/spark-shell \
  --master k8s://https://aksspark00-akssparkrg-53a326-07f35845.hcp.southeastasia.azmk8s.io:443 \
  --name spark-shell \
  --deploy-mode client \
  --conf spark.executor.instances=3 \
  --conf spark.kubernetes.executor.limit.cores=1 \
  --conf spark.kubernetes.container.image=akssparkacr0001.azurecr.io/spark:v3

进入 Spark Shell Scala 命令行交互界面

Scala 验证

1 2	scala> sc.parallelize(1 to 100000).count res10: Long = 100000

具体的分析的文件为示例 csv, 名为 diamonds.csv, 并已经提前上传到 Azure Blob Container 名为 sparkshell 中.

# Azure Blob Storage Account 认证
scala> spark.conf.set(
     |     "fs.azure.account.key.akssparksa0001.blob.core.windows.net",
     |     "gWfl5JezYzXs1ub572KZDAWTR9buVb/pb/dHj/iqsKV07fQKSl+JzUqcLWgx4Qr7xQTVPBSVsXhO6Aja/torAw==")

# 加载 blob 文件
scala> val diamonds = spark.read.format("csv").option("header", "true").option("inferSchema", "true").load("wasbs://sparkshell@akssparksa0001.blob.core.windows.net/diamonds.csv")
diamonds: org.apache.spark.sql.DataFrame = [_c0: int, carat: double ... 9 more fields]

# 显示字段类型
scala> diamonds.printSchema()
root
 |-- _c0: integer (nullable = true)
 |-- carat: double (nullable = true)
 |-- cut: string (nullable = true)
 |-- color: string (nullable = true)
 |-- clarity: string (nullable = true)
 |-- depth: double (nullable = true)
 |-- table: double (nullable = true)
 |-- price: integer (nullable = true)
 |-- x: double (nullable = true)
 |-- y: double (nullable = true)
 |-- z: double (nullable = true)

 # 显示前 10 行数据
scala> diamonds.show(10)
+---+-----+---------+-----+-------+-----+-----+-----+----+----+----+            
|_c0|carat|      cut|color|clarity|depth|table|price|   x|   y|   z|
+---+-----+---------+-----+-------+-----+-----+-----+----+----+----+
|  1| 0.23|    Ideal|    E|    SI2| 61.5| 55.0|  326|3.95|3.98|2.43|
|  2| 0.21|  Premium|    E|    SI1| 59.8| 61.0|  326|3.89|3.84|2.31|
|  3| 0.23|     Good|    E|    VS1| 56.9| 65.0|  327|4.05|4.07|2.31|
|  4| 0.29|  Premium|    I|    VS2| 62.4| 58.0|  334| 4.2|4.23|2.63|
|  5| 0.31|     Good|    J|    SI2| 63.3| 58.0|  335|4.34|4.35|2.75|
|  6| 0.24|Very Good|    J|   VVS2| 62.8| 57.0|  336|3.94|3.96|2.48|
|  7| 0.24|Very Good|    I|   VVS1| 62.3| 57.0|  336|3.95|3.98|2.47|
|  8| 0.26|Very Good|    H|    SI1| 61.9| 55.0|  337|4.07|4.11|2.53|
|  9| 0.22|     Fair|    E|    VS2| 65.1| 61.0|  337|3.87|3.78|2.49|
| 10| 0.23|Very Good|    H|    VS1| 59.4| 61.0|  338| 4.0|4.05|2.39|
+---+-----+---------+-----+-------+-----+-----+-----+----+----+----+
only showing top 10 rows

# 提取前 10 行并写入新文件
scala> diamonds.limit(10).write.format("csv").save("wasbs://sparkshell@akssparksa0001.blob.core.windows.net/1.csv")
scala> spark.read.csv("wasbs://sparkshell@akssparksa0001.blob.core.windows.net/1.csv").show()
+---+----+---------+---+----+----+----+---+----+----+----+
|_c0| _c1|      _c2|_c3| _c4| _c5| _c6|_c7| _c8| _c9|_c10|
+---+----+---------+---+----+----+----+---+----+----+----+
|  1|0.23|    Ideal|  E| SI2|61.5|55.0|326|3.95|3.98|2.43|
|  2|0.21|  Premium|  E| SI1|59.8|61.0|326|3.89|3.84|2.31|
|  3|0.23|     Good|  E| VS1|56.9|65.0|327|4.05|4.07|2.31|
|  4|0.29|  Premium|  I| VS2|62.4|58.0|334| 4.2|4.23|2.63|
|  5|0.31|     Good|  J| SI2|63.3|58.0|335|4.34|4.35|2.75|
|  6|0.24|Very Good|  J|VVS2|62.8|57.0|336|3.94|3.96|2.48|
|  7|0.24|Very Good|  I|VVS1|62.3|57.0|336|3.95|3.98|2.47|
|  8|0.26|Very Good|  H| SI1|61.9|55.0|337|4.07|4.11|2.53|
|  9|0.22|     Fair|  E| VS2|65.1|61.0|337|3.87|3.78|2.49|
| 10|0.23|Very Good|  H| VS1|59.4|61.0|338| 4.0|4.05|2.39|
+---+----+---------+---+----+----+----+---+----+----+----+

另外, 跨 Azure Blob Container 的文件读写也可以.

3. 总结

通过以上的几个步骤, 可以实现在 Azure Kubernetes Service 集群上运行 Spark 作业的平台部署. 与传统的 YARN 来做资源管理调度相比较而言更为轻量快捷, 同时可以与 Azure Blob 做集成, 数据存储在 Azure Blob 中进行分析, 既可以保证数据高可靠性也能够实现计算与存储的分离从而提高灵活性.

2019-01-18

All Blogs►Linux►Windows 10►WSL

Win10强大的WSL(Windows Subsystem for Linux)

1. 前言

Hello，大家好！第一篇正式的博客，我们先来说说 WSL。WSL可以让开发者在 Windows 10 下通过 Bash Shell 运行原生的 Ubuntu 用户态二进制程序，工程师们不用再苦恼所用的 Windows 平台上没有合适的 Linux 工具和库了。对于 Linux 的重度用户来说，福音来了。之所以第一篇技术博客介绍 WSL，也是因为我未来技术博客中的实验环境都会以此为基础。那还等什么，让我们来动手实战一下，感受一下 WSL 的强大吧！

2. WSL简介和原理

WSL 相关代码早在 2016 年 1 月下旬便被微软悄悄内置进了 Windows 10 Build 14251 预览版中，此后微软的开发人员制订了 lxcore.sys 与 lxss.sys 这两个新的子系统文件，让其成为 Windows 程序员开发 Linux 应用程序的桥梁。WSL 是由 Windows 内核团队与 Canonical 合作设计和开发的，可以让 Windows 10 下的开发者们在拥有 Windows 中那些强力支持之外，还能使用 Linux 下丰富的开发环境与工具而不用启动另外的操作系统或者使用虚拟机。这绝对是一个“来自开发者，服务开发者”的 Windows 10 特色，它的目的是让开发者们每天的开发工作都变得顺畅而便捷。我们先来看一下 WSL 的架构和原理图：

WSL 对于 Windows 系统来说属于用户态程序，通过虚拟文件系统接口，以 DriveFs 文件系统挂载到 Windows 从而提供和 Windows 的互操作能力。 lxss.sys 和 lxcore.sys 这两个驱动负责模拟 Linux 内核并实时拦截系统调用。相应的驱动会将 Linux 内核调用映射为对应的 Windows 内核调用。根据从微软内部的压力测试工具据来看，WSL 的性能表现非常接近用相同硬件直接运行 Linux 的性能，几乎可以获得同等的 CPU、内存和 I/O 性能，这证明 WSL 在性能方面的表现很出色。

3. WSL配置

3.1 启用 WSL Feature

顺序：Windows 设置 -> 应用和功能 -> 右侧的程序和功能 -> 启动或关闭windows功能 -> 勾选适用于 Linux 的 Windows 子系统

3.2 安装 WSL

Microsoft store 提供了很多 Linux 发行版本可供选择，用户可以根据自己的爱好和习惯去选择自己喜欢的 Linux 发行版本。本文选择 Ubuntu 进行安装，点击启动，第一次会进行初始化安装。

初始化安装完成，需要设置帐号密码，此处可以根据用户习惯去设置，直接使用 root 或者配置其他用户 sudo 免密切换 root，本文选择使用 sudo 模式。

3.3 配置 WSL

具体的使用方式和 Ubuntu Linux 一样，可以选择启动Ubuntu Bash Terminal来使用，或者可以将 Bash 放到后台，通过 Termius SSH 远程连接到 Ubuntu，本文即采用该模式。

3.3.1 配置sudo免密登陆

1
2
3

$ sudo su -
[sudo] password for adminuser:     ##### input password once #####
$ echo "adminuser ALL=(ALL) NOPASSWD: ALL" >> /etc/sudoers

3.3.2 重装openssh并启动

$ apt-get remove openssh-server
$ apt-get update     ##### 默认源 ubuntu canonical, 如果无法访问，可以根据网络状况自行修改或者科学上网 #####
$ apt-get install openssh-server
$ vi /etc/ssh/sshd_config
  修改 PasswordAuthentication no 为 PasswordAuthentication yes
  :wq 保存退出
$ service ssh --full-restart

3.3.3 通过Termius远程连接

Termius 是 Windows 10 上远程连接 Linux 的终端工具，Microsoft store 可以下载，UI、功能性都不错，可以一用。如果大家有自己喜爱的工具，也可以选择喜欢的工具。

3.4 Bash放到后台，SSH服务开机启动

现在有一个问题，每次 Windows 关机开机之后都需要手动启动 Bash.exe，然后 start ssh service 才可以远程连接使用。每次都需要这样做一次，很繁琐。下面就来介绍一下 SSH 服务如何开机启动吧~

3.4.1 sudo免密登陆

根据目前的操作系统的用户，先配置免密登陆，具体参考 3.3.1。

3.4.1 通过VB script开机启动SSH

写一个 VB 通过调用 Bash.exe 启动 SSH 服务的脚本，例如 startssh.vbs：

1
2
3

set ws=wscript.createobject("wscript.shell")
ws.run "C:\Windows\System32\bash.exe",0
ws.run "C:\Windows\System32\bash.exe  -c 'sudo /usr/sbin/service ssh start'",0

然后将startssh.vbs放到如下目录(需要管理员权限)。

1	%AppData%\Microsoft\Windows\Start Menu\Programs\Startup

开机启动配置完毕。

4. 总结

总体上，WSL的配置就结束了。现在可以在此基础上去配置所需要的开发或者其他组件了，其实WSL也可以像虚拟机一样去备份、回滚等等，很多的其他的功能或玩法，有待大家自己开发了~

1. 前言

2. Azure Database Migration Service 介绍

3. Aliyun RDS MySQL 迁移至 Azure Mooncake MySQL PaaS 验证

3.1 实验场景以及前期准备

3.2 Aliyun RDS MySQL 迁移到 Azure Mooncake MySQL PaaS (中国区)

3.2.1 Azure Mooncake 资源创建初始化

3.2.2 创建 Aliyun RDS MySQL 实例

3.2.3 导入示例数据库到 Aliyun RDS MySQL 实例

3.2.4 创建 Azure DMS 实例

3.2.5 在目标 Azure MySQL PaaS 上创建 Schema

3.2.5 创建 Migration Project 并做 Online Migration

4. 总结

1. 前言

2. Apahce Flink 介绍

3. Flink on AKS 实现

3.1 创建 AKS 集群、ACR 及跳板机

3.2 获取 AKS Credential 并查看 AKS Cluster 信息：

3.3 Docker Login 登陆 ACR (Azure Container Registry)

3.4 Flink Streaming on AKS

3.4.1 Streaming 任务示例

3.4.2 构建自定义 Flink Docker 镜像并推送至 ACR

3.4.3 构建 Flink Streaming JobManager 和 Service

3.4.4 构建 Flink Streaming TaskManager

3.4.5 Flink Streaming Job 运行和结果

3.5 Flink Batch on AKS

3.5.1 下载 Flink 二进制文件

3.5.1 编写 ConfigMap Yaml 文件

3.5.2 编写 JobManager Yaml 文件

3.5.3 编写 TaskManager Yaml 文件

3.5.4 编写 JobManager Service Yaml 文件

3.5.5 启动 Flink UI ：

3.5.6 提交 Flink Batch Job 到 AKS 集群 ：

4. 总结

1. Azure Kubernetes Service 及 Apache Spark 简介

2. Apache Spark on AKS 实现

2.1 创建 AKS 集群

2.1.1 创建资源组

2.1.2 创建 3 节点 AKS 集群

2.1.3 获取 AKS Credential:

2.2 在 AKS 集群中创建 Spark 服务帐户并绑定权限

2.2.1 创建 Spark 服务账号

2.2.2 权限绑定

2.3 下载 Spark 二进制文件

2.4 确认 Spark Jar 和 Kubernetes 的版本兼容性

2.5 创建 ACR (Azure Container Registry) 并与 AKS 集成

2.5.1 创建 ACR

2.5.2 ACR 与 AKS 集群集成

2.5.3 登录 ACR

2.6 构建 Spark 镜像

2.6.1 构建并推送镜像至 ACR

2.6.2 Azure 查看docker image:

2.7 运行作业 SparkPi

2.7.1 创建 Storage Account 并将 spark-examples_2.11-2.4.4.jar 上传至 Azure Blob

2.7.2 运行 SparkPi

2.7.3 查看运行结果

2.8 使用 Spark Shell 进行交互式分析 Azure Blob 数据

2.8.1 Docker Image 新打包 v2 版本

2.8.2 使用 Spark Shell 交互式分析 Azure Blob 中的数据

3. 总结

1. 前言

2. WSL简介和原理

3. WSL配置

3.1 启用 WSL Feature

3.2 安装 WSL

3.3 配置 WSL

3.3.1 配置sudo免密登陆

3.3.2 重装openssh并启动

3.3.3 通过Termius远程连接

3.4 Bash放到后台，SSH服务开机启动

3.4.1 sudo免密登陆

3.4.1 通过VB script开机启动SSH

4. 总结

3.5.6 提交 Flink Batch Job 到 AKS 集群：