无标题文章

作者: 哟桑_6481 | 来源:发表于2019-11-29 15:12 被阅读0次

# 云计算实验

如果说亚马逊的AWS（Amazon Web Service）是一个IaaS平台，为用户提供计算服务（EC2）、存储服务（S3）等基础设施服务的话，可以说谷歌的GAE（Google App Engine）就是一个PaaS平台，在GAE上，用户可以构建和运行可扩展的Web和移动应用程序，而无需考虑底层基础设施的部署、维护和扩展。这使得开发人员能够更加专注于业务逻辑，以便快速构建可扩展的应用程序。在IaaS平台上，开发人员需要自己购买虚拟机、存储设备、网络设备等基础设施，并在这些基础设置之上安装和维护操作系统、运行环境和应用程序。尽管这些基础设施大多数已经实现了虚拟化，与传统的物理设施相比已经节省了大量的人力和物力，但维护这些虚拟的基础设施仍然需要相关的专业技术人员，投入大量的时间和精力才能完成。这对于初创型公式或普通的开发人员而言是难于承担的。作为一个PaaS平台，GAE可以让开发人员直接在Google的基础架构上运行网络应用程序。在 GAE之上易构建和维护应用程序，并且应用程序可根据访问量和数据存储需要的增长轻松进行扩展。使用GAE，开发人员将不再需要维护服务器或虚拟机，只需上传应用程序，它便可立即即为用户提供服务。GAE是一个商业化的PaaS平台，自2008 年 4月发布第一个版本以来，已吸引了全球数十万计的开发者在其上开发了众多的应用程序。

作为谷歌推出的一款商业化云计算平台，GAE的底层是建立在Google的基础架构之上的，开发人员和用户在选择GAE的同时，也就选择了Google提供基础设施资源，这在一定程度上限制了开发人员和用户的可选择性。相对而言，另一个由社区发起的开源PaaS项目——AppScale，在这方面可以更好的适应开发人员和用户的需求。AppScale 是GAE的开源实现，它同时也是一个开源的PaaS平台，允许用户在任何地方（服务器或集群）发布和托管自己的 GAE 应用程序。 AppScale支持 GAE平台的Python、Java、PHP和Go的运行时库。用户可以无缝地将应用程序在公共云和自己的虚拟机、私有云、GAE或者其他云平台（如AWS）环境中移植。由于是开源软件，AppScale不依赖于任何特定厂商的基础设施，开发人员既可以根据实际情况选择不同厂商提供的基础设施服务，也可以根据需求从头搭建属于自己的基础设施环境。

接下来，我们就以AWS作为IaaS平台，以AppScale作为PaaS平台，通过几个相互关联的实验，展示云应用、大数据和人工智能领域的几个典型应用案例的开发过程。

## 第1节云应用实验

云应用是面向“云”而设计的应用，因此技术部分依赖于在传统云计算的3层概念（基础设施即服务（IaaS）、平台即服务（PaaS）和软件即服务（SaaS））。云应用是天然适合云特点的应用，云原生应用系统需要与操作系统等基础设施分离，不应该依赖Linux或Windows等底层平台，或依赖某个云平台。也就是说，应用从开始就设计为运行在云中，无论私有云或公有云；其次，该应用必须能满足扩展性需求，垂直扩展（向上和向下）或水平扩展（跨节点服务器）。

由于云应用程序开发采用与传统本地应用程序完全不同的体系结构，云应用和本地应用之间存在着较大的差异。例如，在可扩展性方面，传统的本地应用面向的用户和对资源的消耗量相对稳定，对应用动态扩展的要求不高。而在云环境下，云应用需要面对大量的互联网用户，用户群体巨大，而访问时间和对资源的消耗量不够确定，有较大的随机性。通常情况下，云应用程序可利用云的弹性，在峰值期间动态增加资源，峰值消退后动态释放资源。云应用根据需要调整资源和规模，既动态适应需求变化，又避免资源浪费。在编程语言方面，编写在服务器上运行的本地部署应用程序往往使用传统语言编写，如C/C ++，C＃或其他Visual Studio语言（如果部署在Windows Server平台上）和企业级Java。对于云应用的开发，面对的问题和采取的技术与传统本地应用开发存在较大的差异，云应用更多是采用以网络为中心的语言编写，这意味着使用HTML，CSS，Java，JavaScript，.Net，Go，Node.js，PHP，Python和Ruby等。接下来，我们先介绍AppScale云应用开发环境的部署方法，然后通过几个实验分别展示使用Python、PHP、Go和Java语言开发云应用的方法。

AWS上提供了预装的AppScale环境，下面介绍在AWS云平台上启动预装AppScale环境的全过程。

* 使用账号密码登录AWS管理控制台，然后选择‘EC2’

![alt picture](./images/appscale/1.png)

* 继续点击‘启动实例’

![alt picture](./images/appscale/2.png)

步骤1：选择AMI。点击‘AWS Marketplace’，然后搜索appscale，点击‘选择’，继续点击‘Continue’

![alt picture](./images/appscale/3.png)

步骤2~6一直保持默认，选择‘下一步’，直到步骤7，点击‘启动’

![alt picture](./images/appscale/10.png)

选择‘创建新密钥对’，并输入密钥对名称，然后下载密钥对保存到本地，继续点击‘启动实例’

![alt picture](./images/appscale/11.png)

* 点击‘查看实例’，选中刚刚新建的实例，可以在下方观察到该实例的域名

![alt picture](./images/appscale/13.png)

打开远程登录工具Xshell，输入``ssh -i "密钥对文件地址" ubuntu@AWS实例域名``，选择之前下载的密钥对文件，并填入密钥对名称，进行远程登录

注意：若无法连接，可能是远程连接端口22未打开，点击下方的‘查看入站规则’进行检查，若没有22端口，则需要点击左侧链接，新建22端口的入站规则即可。

![alt picture](./images/appscale/14.png)

切换为root身份并进入/root家目录：``sudo su`` -> ``cd ~``

查看AppScale状态：``appscale status``

![alt picture](./images/appscale/15.png)

测试：打开浏览器输入``http://域名:1080``或者``https://域名:1443``

![alt picture](./images/appscale/启动测试.png)

如果无法使用默认账号和密码登录系统，可以查看AppScale的配置文件AppScalefile，该文件默认存放在/root/目录中，可通过vim命令查看配置文件AppScalefile的内容：``vim AppScalefile``

![alt picture](./images/appscale/配置文件.png)

AppScale配备了一套名为appscale-tools的工具，可以对AppScale平台执行基本的管理任务。例如：

查看AppScale状态：``appscale status``

启动AppScale：``appscale up``

关闭AppScale：``appscale down``

收集AppScale日志：``appscale logs 路径``

登录AppScale：``appscale ssh``

升级AppScale：``appscale upgrade``

### 实验1. 欢迎页面

__实验内容：分别使用Python、PHP和Go语言，在AppScale平台上开发一个Web应用，该应用在浏览器中显示一个非常简单的欢迎页面。__

首先使用Python语言进行开发。为了完成Python版本的Hello程序，我们需要准备两个文件：app.yaml 和 hello.py。第一个文件 app.yaml 是应用的配置文件，文件内容如下：

```

application: hello-python

runtime: python27

api_version: 1

threadsafe: true

handlers:

- url: /.*

script: hello.app

```

其中，`application: hello-python` 表示应用名称，该名称会显示在AppScale的应用列表中。`runtime: python27` 表示该应用在python 2.7环境中运行。`url: /.*` 中包含了一个url的通配符，表示客户向任何 url 路径发起的请求，都会转交给 `script: hello.app` 中指定的 `hello.app` 处理。hello.app 是一个处理客户端请求的对象，它在第二个文件 hello.py 中定义，hello.py 是一个用 Python 语言编写的源程序文件，文件内容如下：

```

import webapp2

class MainPage(webapp2.RequestHandler):

def get(self):

self.response.out.write("<h1 style='text-align: center;'>Hello Python</h1><p style='text-align: center;'>Welcom to Cloud APP World!</p>")

app = webapp2.WSGIApplication([('/', MainPage),], debug=True)

```

其中，`app = webapp2.WSGIApplication` 一行定义了处理网络请求的变量app，该变量被实例化为MainPage类的一个对象。在类MainPage中定义了get方法，该方法对所有网络请求进行回应，回应中包含了向客户端浏览器输出的 HTTP 协议头和 HTML 格式的文本信息：`<h1 style='text-align: center;'>Hello Python</h1><p style='text-align: center;'>Welcom to Cloud APP World!</p>` ，这些信息最终会以一个 Web 页面的形式在客户端浏览器中显示出来。

准备好 app.yaml 和 hello.py 两个文件后，需要将它们上传到 AppScale 平台。先以管理员的账号登录AppScale，点击页面左侧导航栏中的 Upload Application，进入应用上传页面，如图0所示。

<br>图0. 应用上传页面<br></div>

从页面提示中可以看出，要将应用上传到AppScale平台，必须先将该应用的全部文件打包成一个压缩文件，然后再上传该压缩文件。对于Python、Go和PHP语言编写的应用，压缩文件中必须包含该应用的配置文件app.yaml。按照AppScale的要求，将 app.yaml 和 hello.py 两个文件打包成zip格式的压缩文件 package.zip，如图0所示。

点击应用上传页面中的“选择文件”按钮，选择 package.zip 文件，如图0所示。

点击 Upload 按钮，上传文件，如图0所示。

文件上传成功后，页面会出现应用上传成功的提示，如图0所示。

<br>图0. 应用上传成功<br></div>

回到AppScale首页，可以看到正在运行的应用列表中，出现了刚才新上传的应用 hello-python，如图0所示。

<br>图0. Hello Python应用入口<br></div>

图中可以看到，跟应用 hello-python 相关的链接有两个，其中，前一个为 http 链接，后一个为更加安全的 https 链接。点击其中任何一个链接，都可以打开该应用。应用的运行效果如图0所示。

<br>图0. Hello Python应用<br></div>

至此，我们就完成了AppScale上的第一个Python应用：Hello Python 的开发和部署过程。

接下来使用PHP语言进行开发。为了完成PHP版本的Hello程序，我们需要准备两个文件：app.yaml 和 hello.php。第一个文件 app.yaml 是应用的配置文件，文件内容如下：

```

application: hello-php

version: 1

runtime: php

api_version: 1

handlers:

- url: /.*

script: hello.php

```

其中，`application: hello-php` 表示应用名称，该名称会显示在AppScale的应用列表中。`runtime: php` 表示该应用在php环境中运行。`url: /.*` 中包含了一个url的通配符，表示客户向任何 url 路径发起的请求，都会转交给 `script: hello.php` 处理。hello.php 是需要我们准备的第二个文件，是一个用 PHP 语言编写的源程序文件，文件内容如下：

```

<?php

echo "<h1 style='text-align: center;'>Hello PHP</h1><p style='text-align: center;'>Welcom to Cloud APP World!</p>";

```

该文件内容非常简单，就是向客户端浏览器输出的一段 HTML 格式的文本信息：`<h1 style='text-align: center;'>Hello PHP</h1><p style='text-align: center;'>Welcom to Cloud APP World!</p>` ，这些信息最终会以一个 Web 页面的形式在客户端浏览器中显示出来。

接下来将 app.yaml 和 hello.php 两个文件打包成zip格式的压缩文件 package.zip，如图0所示。

通过应用上传页面上传 package.zip 文件，如图0所示。

文件上传成功后，回到AppScale首页，可以看到正在运行的应用列表中，出现了刚才新上传的 hello-php 应用，如图0所示。

<br>图0. Hello PHP应用入口<br></div>

图中可以看到，跟应用 hello-php 相关的链接有两个，其中，前一个为 http 链接，后一个为更加安全的 https 链接。点击其中任何一个链接，都可以打开该应用。应用的运行效果如图0所示。

<br>图0. Hello PHP应用<br></div>

至此，我们就完成了AppScale上的第一个PHP应用：Hello PHP的开发和部署过程。

接下来使用Go语言进行开发。为了完成Go版本的Hello程序，我们需要准备两个文件：app.yaml 和 hello.go。第一个文件 app.yaml 是应用的配置文件，文件内容如下：

```

application: hello-go

version: 1

runtime: go

api_version: go1

handlers:

- url: /.*

script: hello.go

```

其中，`application: hello-go` 表示应用名称，该名称会显示在AppScale的应用列表中。`runtime: go` 表示该应用在go环境中运行。`url: /.*` 中包含了一个url的通配符，表示客户向任何 url 路径发起的请求，都会转交给 `script: hello.go` 处理。hello.go 是需要我们准备的第二个文件，是一个用 Go 语言编写的源程序文件，文件内容如下：

```

package hello

import (

"fmt"

"net/http"

)

func init() {

http.HandleFunc("/", main)

}

func main(w http.ResponseWriter, r *http.Request) {

fmt.Fprintf(w, "<h1 style='text-align: center;'>Hello Go</h1><p style='text-align: center;'>Welcom to Cloud APP World!</p>");

}

```

其中，`func main(w http.ResponseWriter, r *http.Request)` 一行定义了程序执行的主函数main，该函数对所有网络请求进行回应，回应中包含了向客户端浏览器输出的一段 HTML 格式的文本信息：`<h1 style='text-align: center;'>Hello Go</h1><p style='text-align: center;'>Welcom to Cloud APP World!</p>` ，这些信息最终会以一个 Web 页面的形式在客户端浏览器中显示出来。

接下来将 app.yaml 和 hello.go 两个文件打包成zip格式的压缩文件 package.zip，如图0所示。

通过应用上传页面上传 package.zip 文件，如图0所示。

文件上传成功后，回到AppScale首页，可以看到正在运行的应用列表中，出现了刚才新上传的 hello-go 应用，如图0所示。

<br>图0. Hello Go应用入口<br></div>

图中可以看到，跟应用 hello-go 相关的链接有两个，其中，前一个为 http 链接，后一个为更加安全的 https 链接。点击其中任何一个链接，都可以打开该应用。应用的运行效果如图0所示。

<br>图0. Hello Go应用<br></div>

至此，我们就完成了AppScale上的第一个Go应用：Hello Go的开发和部署过程。

最后，我们尝试将本实验上传的3个应用从AppScale平台删除。

点击AppScale页面左侧导航栏中的 Delete Application，进入删除应用页面，点击页面中选择应用的下拉框，如图0所示。

<br>图0. 删除应用页面<br></div>

从下拉框中分别选择 hello-python、hello-php和hello-go应用，依次删除，如图0所示。

三个应用都删除完毕后，回到AppScale首页，可以看到，正在运行的应用列表中已经没有hello-python、hello-php和hello-go三个应用的信息了，如图0所示。

<br>图0. Hello Go应用入口<br></div>

至此，我们就完成了AppScale上的三个版本的Hello应用：hello-python、hello-php和hello-go的开发、部署和删除过程。

### 实验2. 访客留言

__实验内容：使用Python语言，在AppScale平台上开发一个记录访客留言的应用。访客可以在应用中发布自己的留言，也可以查看其他访客发布的留言。__

在AppScale的官方安装包中有一个GuestBook应用，该应用使用Python语言开发，实现了最基本的留言簿功能。GuestBook的实现过程采用了AppScale提供的几个关键技术，如：网站框架、用户认证和数据存储等。这些技术几乎在所有AppScale应用中都会被用到，所以理解和掌握GuestBook应用的实现过程，是学习AppScale应用开发的一个很好的起点。

在正式分析GuestBook代码之前，先了解一下程序的运行效果。由于是预装应用，默认情况下，在AppScale首页的应用列表中就可以看到GuestBook应用入口，如图0所示。

<br>图0. GuestBook应用入口<br></div>

点击两个链接中的任意一个，即可进入GuestBook应用界面，如图0所示。

<br>图0. GuestBook应用界面<br></div>

应用页面，会显示最新发布的10条留言。因为是第一次打开应用，所以当前页面上还没有任何留言信息。现在我们在页面输入框中输入一个新的留言`Hello, GuestBook!`，点击`Sign Guestbook`按钮，该留言就会在页面上显示出来，留言者为当前账户：a@a.com，如图0所示。

<br>图0. 第一条留言信息<br></div>

点击页面右上角的Logout按钮可退出当前用户，此时，Logout按钮会变成Login按钮。当前用户退出后，仍然可以在页面中发布新的留言。例如，输入一个新的留言`Who am I?` 点击`Sign Guestbook`按钮，该留言会在页面顶端显示出来，只是相应的留言者变成了匿名用户` An anonymous person `，如图0所示。

点击页面右上角的Login按钮，会进入AppScale的登录页面。如图0所示。

在图0所示的登录页面中，既可以使用已有AppScale账户登录，也可以点击`Create one` 进入创建用户页面，在该页面中创建新的AppScale账户。例如，创建新用户b@b.com，如图0所示。

<br>图0. 创建新用户<br></div>

新用户创建成功后，点击`Login` 按钮返回登录页面，使用新用户登录，如图0所示。

<br>图0. 新用户登录<br></div>

登录成功后，页面会跳转到GuestBook的应用页面，页面右上角的Logout按钮表示，已经有用户登录。再尝试发布一个留言`Hello, I'm Back! `，可以看到，对应的留言者为刚才新登录的用户：b@b.com。如图0所示。

<br>图0. 新用户登录<br></div>

需要特别注意的是，图0所示的登录界面和图0所示的创建用户界面并不仅仅属于GuestBook应用，它们是AppScale提供的统一的身份认证模块，属于整个AppScale平台。运行在AppScale平台上的任何应用，都可以直接使用这个模块，无需从头开发。当然这并不是必须的，在大多数情况下，使用AppScale提供的统一身份认证模块，可以有效减少应用开发工作量，还可以更好的与其他应用交互，提高应用的适应性和扩展性。

上述步骤展示了GuestBook应用的基本功能。接下来分析GuestBook应用的实现方法。

当用户点击GuestBook应用入口（图0），准备进入GuestBook应用页面时，用户端浏览器会向AppScale服务器发起页面请求。AppScale服务器收到用户请求后，会根据请求页面的url路径，找到对应的处理程序，并将请求转发给该处理程序。请求页面的url路径和处理程序之间的对应关系，在应用配置文件app.yaml中指定，文件内容如下：

```

application: guestbook

runtime: python27

api_version: 1

threadsafe: true

handlers:

- url: /favicon\.ico

static_files: favicon.ico

upload: favicon\.ico

- url: /bootstrap

static_dir: bootstrap

- url: /.*

script: guestbook.application

libraries:

- name: webapp2

version: latest

- name: jinja2

version: latest

```

第一次打开GuestBook应用的网页请求地址为：`/`，和app.yaml文件中`url: /.*`一行匹配，对应的处理程序为`script: guestbook.application`，它表示guestbook.py文件中定义的application对象。接下来在guestbook.py文件中查看application对象的定义，代码如下：

```

application = webapp2.WSGIApplication([

('/', MainPage),

('/wait', WaitHandler),

('/sign', Guestbook),

], debug=True)

```

该定义表示，application是一个处理网页请求的对象，它会根据请求地址的不同，将请求转发给不同的处理程序。此次请求地址为：`/`，对应的处理程序为MainPage，代码如下：

```

class MainPage(webapp2.RequestHandler):

def get(self):

greetings_query = Greeting.query(ancestor=guestbook_key()).order(-Greeting.date)

greetings = greetings_query.fetch(10)

if users.get_current_user():

url = users.create_logout_url(self.request.uri)

url_linktext = 'Logout'

else:

url = users.create_login_url(self.request.uri)

url_linktext = 'Login'

template = jinja_environment.get_template('index.html')

self.response.out.write(template.render(greetings=greetings,

url=url,

url_linktext=url_linktext))

```

其中，Greeting是一个新定义的数据库实体类型，用于记录用户的留言信息。每一个Greeting实体包含3个字段：auther、content和data，分别记录一条留言的作者、内容和时间。Greeting继承自AppScale内置的数据库实体类型，定义如下：

```

class Greeting(ndb.Model):

author = ndb.UserProperty()

content = ndb.StringProperty(indexed=False)

date = ndb.DateTimeProperty(auto_now_add=True)

```

接下来定义一个名为`default_guestbook`的留言本（Guestbook实体），它是所有留言信息（Greeting实体）的父节点。定义一个函数guestbook_key() 来访问这个留言本，定义如下：

```

def guestbook_key(guestbook_name='default_guestbook'):

return ndb.Key('Guestbook', guestbook_name)

```

回到MainPage中的下述代码位置：

`greetings_query = Greeting.query(ancestor=guestbook_key()).order(-Greeting.date)`

这行代码首先调用`query(ancestor=guestbook_key())`，获取祖先节点为`default_guestbook`留言本的所有留言信息，再调用`order(-Greeting.date)`函数对这些留言信息按发布时间进行排序，排序后的结果存放到`greetings_query`变量中。接下来的代码：

`greetings = greetings_query.fetch(10)`

从`greetings_query`中获取前10条留言信息，即最新发布的10条信息，存放到`greetings`变量中。

接下来的代码：

`users.get_current_user()`

会获取AppScale的登录账号，如果获取成功，表示当前已有用户登录，则生成退出登录的按钮：

```

url = users.create_logout_url(self.request.uri)

url_linktext = 'Logout'

```

如果获取失败，表示当前没有用户登录，则生成登录系统的按钮：

```

url = users.create_login_url(self.request.uri)

url_linktext = 'Login'

```

接下来输出页面显示的内容：

```

template = jinja_environment.get_template('index.html')

self.response.out.write(template.render(greetings=greetings,

url=url,

url_linktext=url_linktext))

```

先将模板文件index.html加载到变量`template` 中，再将最新留言（greetings）、登录按钮（Logint）或退出按钮（Logout）信息导入变量`template`，最后调用write函数将页面内容输出到客户端浏览器。其中，模板文件index.html中包含了输出页面内容的代码，如显示登录/退出按钮的代码如下：

`<a href="{{ url|safe }}" class="btn">{{ url_linktext }}</a>`

其中的url和url_linktext的值，就来自于上述MainPage代码中给的url和url_linktext变量赋值的内容。

index.html文件中显示最新留言信息的代码如下：

```

{% for greeting in greetings %}

{% if greeting.author %}

<b>{{ greeting.author.nickname() }}</b> wrote:

{% else %}

An anonymous person wrote:

{% endif %}

<blockquote>{{ greeting.content }}</blockquote>

</div>

{% endfor %}

```

该代码会轮询待显示的全部留言（greetings），逐一输出每一条留言（greeting）的内容（greeting.content）和留言者（greeting.author）。从代码中可以看出，如果已有用户登录，则greeting.author不为空，留言者位置会显示当前用户；如果没有用户登录，则greeting.author为空，留言者位置会显示`An anonymous person`。

index.html文件最后显示的内容是用户输入留言的表单，见图0中C所示的区域。

<br>图0. GuestBook界面<br></div>

显示该表单的代码如下：

```

</form>

```

其中，`<textarea name="content" class="input-block-level" rows="3"></textarea>`对应一个3行的文本框，用户在这个文本框中输入留言信息，这些信息将被存放在content字段中，待提交表单时传递给服务器。`<form action="/sign" method="post">`表示提交表单时，表单中的信息会以post方式传递到服务器，并请求地址为`/sign`的页面处理该信息。回到application对象的定义，代码如下：

```

application = webapp2.WSGIApplication([

('/', MainPage),

('/wait', WaitHandler),

('/sign', Guestbook),

], debug=True)

```

该定义中` '/sign', Guestbook`一行表示，请求地址为`/sign`时，对应的处理程序为Guestbook，代码如下：

```

class Guestbook(webapp2.RequestHandler):

def post(self):

greeting = Greeting(parent=guestbook_key())

if users.get_current_user():

greeting.author = users.get_current_user()

greeting.content = self.request.get('content')

greeting.put()

self.redirect('/')

```

其中，`post(self)`表示用post方式接收表单数据。`greeting = Greeting(parent=guestbook_key())`定义了一个新的 greeting 变量，用于存放一个新的留言，同时将留言设置为`default_guestbook`留言本的子节点。接下来的两条语句分别设置了greeting变量的author字段和content的字段值，分别代表留言人和留言内容。下一条语句`greeting.put()`将这条留言信息存入AppScale数据库。最后一条语句`self.redirect('/')`，把页面重定向到新的页面：`/`。页面重定相当于向新的页面发起请求，根据application对象的定义：

```

application = webapp2.WSGIApplication([

('/', MainPage),

('/wait', WaitHandler),

('/sign', Guestbook),

], debug=True)

```

此次请求，会再次调用`/`对应的处理程序：MainPage，从而再次进入页面的显示流程。当再次显示页面时，MainPage中的下述代码：

`greetings_query = Greeting.query(ancestor=guestbook_key()).order(-Greeting.date)`

会获取祖先节点为`default_guestbook`留言本的所有留言信息，其中就包括上一步存入数据库的最新留言，最终在页面上显示出来，如图0所示。

至此，我们就完成了AppScale上的GuestBook应用的分析工作。

### 实验3. 应用中心

__实验内容：在AppScale平台上开发云应用中心，对不同类型的云应用进行统一的发布、管理和运维操作。__

AppScale支持使用Python、PHP、Go和Java语言编写Web应用。但对于传统类型的本地应用，即在服务器上部署和运行本地应用程序，AppScale是无法直接执行并将结果显示到客户端浏览器的。然而，传统的本地应用数量巨大，如何才能将这些应用移植到云上，尽量减少代码的修改量，是一个值得思考的问题。本实验尝试使用一种简单且通用性较强的方法，把云应用分为应用中心（APP Center）和应用代理（APP Agent）两个模块。应用中心运行在AppScale服务器上，作为统一管理和调度云应用的控制台。应用代理运行在传统的本地应用所在的服务器上，作为本地应用的实际调用者。采用远程调用的方式，由云应用中心统一向应用代理发起调用请求。应用代理接到请求后，启动本地应用程序。本地应用程序执行的过程中，可以通过应用代理从应用中心获取参数，也可以通过应用代理把程序的执行结果返回给应用中心。应用中心通过应用代理和本地应用程序交互，最终将程序执行状态或执行结果输出到用户端浏览器中。

本实验在实验2的基础上，逐步修改和扩充程序代码，直到完全实现上述功能。

首先，用一个C语言编写的Hello World程序作为本地应用的示例。该程序的执行结果就是输出一个字符串`Hello C`，如图0所示。

图0 Hello C程序

接下来使用Python语言开发应用代理（APP Agent）程序，程序代码如下：

```

#!/usr/bin/python

# -*- coding: UTF-8 -*-

import socket

import os

s=socket.socket()

s.bind(('0.0.0.0',1111))

s.listen(5)

while True:

c,addr = s.accept()

cmd = './' + c.recv(1024)

print(cmd)

result =os.popen(cmd).read()

print(result)

c.send(result)

c.close()

```

该代码引用了socket和os库，前者用于建立与APP Center通信的tcp连接，后者用于调用本地应用程序（如Hello C程序）。Agent程序先定义一个socket对象s，该对象在本机1111端口监听，当接收到远程连接后（由APP Center发起），会新建另一个socket对象c，由c负责从该连接接收命令。接收完毕后，先和'./'组装起来，表示要调用当前目录下的这个命令，然后通过popen函数执行该命令，并将执行结果通过read函数读入到result变量中。最后把result的内容通过send函数发送给远程连接（由APP Center接收）。

接下来将Guestbook程序改造成APP Center程序。先调整访客留言程序的文字和布局，使其更加接近于一个应用控制台的显示效果。在文字方面，把页面顶端的App Engine Guestbook改为Cloud APP Center。把Sign Guestbook改为Run，表示执行应用。把留言人账号后面的wrote改为run，表示某用户执行了某应用。在布局方面，把输入框从页面底部，调整到页面上方，方便用户输入。在本实验中，用户在输入框中输入的信息不再是留言信息，而是用户希望执行的应用程序和相关信息。我们要把这些信息，显示到"用户名 run:"的右边，而把下边的显示区域留出来，作为应用程序执行结果的显示区。调整之后的效果如图0所示。

图0 文字和布局

然后在结果显示区中显示应用程序的执行结果。这一步需要在用户输入希望执行的指令后，建立APP Center与APP Agent的通信，把指令信息发送给APP Agent。待APP Agent调用相应程序后，再将程序执行结果传回APP Center。这一步是在点击Run按钮后，提交表单的程序中实现的，具体代码如下：

```

class Guestbook(webapp2.RequestHandler):

def post(self):

greeting = Greeting(parent=guestbook_key())

if users.get_current_user():

greeting.author = users.get_current_user()

cmd = self.request.get('content')

c = socket.socket()

c.connect(('127.0.0.1',1111))

c.send(cmd)

result = c.recv(1024)

c.close()

greeting.content = cmd

greeting.result = result

greeting.put()

self.redirect('/')

```

其中，`self.request.get('content')`是用户在留言框中提交的信息，在上一个实验中表示留言信息，在本实验中表示用户希望执行的指令，将该信息存入cmd变量。接下来定义一个socket对象c，连接到APP Agent程序所在服务器的1111端口。当前APP Agent程序和APP Center程序运行在同一台服务器上，所以IP地址为`127.0.0.1`。连接建立好后，APP Center先将要执行的指令发送给APP Agent，再从APP Agent获取该指令的执行结果，存放到result变量中。随后，将指令和执行结果通过greeting对象存入数据库中，将网页重定向到根目录对应的页面。从上一个实验的分析可知，根目录对应的页面文件为index.html。为了将指令和执行结果信息在该页面显示出来，需要对index.html文件做如下修改：

```

{% for greeting in greetings %}

{% if greeting.author %}

<b>{{ greeting.author.nickname() }}</b> run: {{ greeting.content }}

{% else %}

An anonymous person run: {{ greeting.content }}

{% endif %}

<blockquote>{{ greeting.result }}</blockquote>

</div>

{% endfor %}

```

其中，`An anonymous person run: {{ greeting.content }}` 一行用于在显示该用户希望执行的指令。`<blockquote>{{ greeting.result }}</blockquote>` 一行用于显示APP Center从APP Agent处获取的执行结果。修改后的显示效果如图0所示。

图0

到目前为止，我们只有一个APP Agent，所以可以将它的IP地址（`127.0.0.1`）固定写在程序源码中，每次用户发起的执行命令，都会发送到该APP Agent执行。换句话说，所有可执行的应用都只能安装在这一台服务器上（`127.0.0.1`）。接下来，我们继续对程序进行扩充，使得被调用的应用程序可以运行在不同的服务器上。最简单的扩充方式，是要求用户在输入待执行指令的时候，在输入信息中增加APP Agent所在的服务器IP地址。例如，输入：`192.168.1.100 hello`，表示希望在`192.168.1.100`服务器上执行hello程序；输入：`192.168.1.200 hello`，表示希望在`192.168.1.200`服务器上执行hello程序。在APP Center处理输入框信息的相应位置，增加提取服务器IP地址的代码，具体代码如下所示：

```

cmd = self.request.get('content')

c = socket.socket()

c.connect(('127.0.0.1',1111))

c.send(cmd)

result = c.recv(1024)

c.close()

```

其中，???表示从输入信息中提取服务器IP地址，？？？表示将剩余的字符串作为待执行的指令信息。其余代码保持不变。为了测试程序运行效果，我们使用了两台APP Agent服务器，IP地址分别为`192.168.1.100 `和`192.168.1.200`。其中，在第一台服务器上安装了hello1程序，程序执行结果为`Hello, I'm No1!`；在第二台服务器上安装了hello2程序，程序执行结果为`Hello, I'm No2!`。通过修改后的APP Center程序分别向两台APP Agent服务器发起程序执行请求，执行结果如图0所示。

图0

从图0中可以看出，只要用户正确指定了服务器IP地址和相应的应用程序，APP Center就能够和相应的APP Agent通信，调用相应的应用程序并获得执行结果。目前的实现方案比较适合于服务器和应用数量较少的情况，一旦规模变大，要用户记住每一个应用所在的服务器IP地址和相应的指令是不现实的。虽然目前程序功能已经实现，但使用起来还不够方便，用户体验还不够好。接下来对程序进行进一步的优化，让用户的使用体验更好。优化的目标有以下几点：

1. 用户不需要记住云平台支持多少应用。

2. 用户不需要记住每个应用服务器的IP地址。

3. 用户不需要记住执行每个应用的详细命令和参数。

所以，对于的开发需求有以下几条：

1. 新增应用登记功能，将云平台支持的应用登记到系统中，使用时可从下拉列表中直接选择应用。

2. 登记应用时，需要同时将应用服务器IP地址登记下来，用户使用应用时不需要重新输入。

3. 如果应用执行时需要参数，则在用户选择该应用后，显示输入参数的提示信息。

首先实现应用登记页面，界面如图0所示。

登记页面用一个表单实现，对应的部分代码如下所示：

```

<form action="/create" method="post">...

<label for="ip">Ip:</label>...

<input id="ip" name="ip" size="40" type="text" value="" />...

<label for="name">Name:</label>...

<input id="name" name="name" size="40" type="text" value="">...

<label for="description">Description:</label>...

<textarea id="description" name="description" class="input-block-level" rows="1" style="width:105%"></textarea>...

<label for="cmd">Cmd:</label>...

<input id="cmd" name="cmd" size="40" type="text" value="">...

<input style="margin-top:10%;width:auto" class="btn btn-primary" name="commit" type="submit" value="注册" />...

</form>

```

点击 `注册` 后，表单数据将以post方式提交到 `/create` 地址。

```

application = webapp2.WSGIApplication([

('/', MainPage),

('/wait', WaitHandler),

('/sign', Guestbook),

('/create', CreatApp),

], debug=True)

```

上述代码将提交到`/create` 地址的数据交给`CreatApp` 对象处理，`CreatApp` 对象的`post`函数负责处理接收到的数据，相关代码如下：

```

class CreatApp(webapp2.RequestHandler):

def post(self):

app = App(parent=APP_key())

app.ip = self.request.get('ip')

app.name = self.request.get('name')

app.description = self.request.get('description')

app.cmd = self.request.get('cmd')

app.put()

self.redirect('/')

```

其中，`app = App(parent=APP_key())`定义了一个新的 app 变量，用于存放新注册的应用信息，同时将该应用的父节点设置为`APP_key()`。接下来的四条语句分别设置了app 变量的ip、name、description和cmd字段，分别代表留应用所在服务器的IP地址、应用名、详细描述和指令。下一条语句`app.put()`将这条留言信息存入AppScale数据库。最后一条语句`self.redirect('/')`，把页面重定向到新的页面：`/`。根据application对象的定义，此次请求，会再次调用`/`对应的处理程序MainPage，从而再次进入主页面的显示流程。

接下来修改主页面的显示程序MainPage，增加如下代码：

```

class MainPage(webapp2.RequestHandler):

def get(self):...

apps_query = App.query(ancestor=APP_key())...

apps = apps_query.fetch()...

template = jinja_environment.get_template('index.html')...

self.response.out.write(template.render(greetings=greetings,

apps=apps,

url=url,

url_linktext=url_linktext))...

```

其中，`apps_query = App.query(ancestor=APP_key())` 一行从AppScale数据库中提取全部父节点为`APP_key()` 的节点信息，这些节点就是在应用注册页面中提交的全部应用信息。`self.response.out.write(...)` 函数中的第二个参数 `apps=apps` ，将这些应用信息传递到`index.html` 页面并显示出来。`index.html` 页面中实现了一个应用选择的下拉框，点击后会显示一个下拉列表，下列列表中的每一行对应一个应用信息，相关代码如下：

```

<form action="/sign" method="post">...

{% for app in apps %}

{% if app.name %}

<p hidden>{{app.description}}</p>

</li>

{% endif %}

{% endfor %}

</ul>

</form>

```

注意，每个应用的注册信息都包含name、ip、cmd、description等多个字段，在下拉列表中，仅需要显示name这个字段即可，其他字段隐藏起来，当用户点击某个应用时，会调用`onclick="shows($(this))` 中指定的shows函数，在该函数中会使用到这些隐藏字段。shows函数中相关代码如下：

```

function shows(a){

$('.buttonText1').text(a.context.children[0].text);

$('.showdescription').text(a.context.children[2].textContent);

$('input#submitIp').val(a.context.children[3].textContent);

$('input#submitCmd').val(a.context.children[4].textContent);

$('input#submitName').val(a.context.children[0].text);

$('.showdescription1').text(a.context.children[0].text+' DESCRIPTION:');

```

其中，`a.context.children[...]`用于访问下列框元素内的子元素，例如，children[0]对应app.name元素，children[1]对应app.id元素，children[2]对应app.description元素。`$('.buttonText1')` 、`$('.showdescription')`、`$('.showdescription1')`对应页面上3个显示区域，`$('.input#submitIp')` 、`$('.input#submitCmd')`、`$('.input#submitName')`对应表单中id分别为`submitIp`、`submitCmd` 和 `submitName` 的3个隐藏的输入框：

```

<form action="/sign" method="post">...

<input type="hidden" id="submitIp" name="submitIp" value="" />...

<input type="hidden" id="submitCmd" name="submitCmd" value="" />...

<input type="hidden" id="submitName" name="submitName" value="" />...

</form>

```

当用户选择一个应用后，页面上3个显示区域会发生相应的变化，其中，应用详情区域会显示应用执行需要的参数信息，如图0所示。

图0

同时，3个隐藏的输入框会分别获取应用的IP、cmd、name信息。用户根据应用详情的提示，在参数输入框中输入相关参数后，点击`RUN`按钮，表单数据将会以post方式提交到`/sign`地址处理。根据application对象的定义，提交到`/sign`地址的数据将会被转交给Guestbook对象处理。接下来修改Guestbook对象的post函数，用于处理接收到的表单数据，相关代码如下：

```

class Guestbook(webapp2.RequestHandler):

def post(self):...

cmd = self.request.get('submitCmd')+' '+self.request.get('canshu')

ip = self.request.get('submitIp')

c = socket.socket()

c.connect((ip,1111))

c.send(cmd)

result = c.recv(1024)

c.close()...

```

在原先程序的基础上，修改了获取程序执行指令、APP Agent所在服务器的IP地址的代码。将用户选择的应用名称和参数信息，组装成完整的程序执行指令，发送给应用所在服务器的APP Agent程序，并从APP Agent程序获取应用执行结果，最后在APP Center页面显示出来。修改后的程序运行效果如图0所示。

图0

为了测试程序运行效果，我们仍然使用之前准备好的两台APP Agent服务器，IP地址分别为`192.168.1.100 `和`192.168.1.200`。其中，第一台服务器上安装了hello1程序，第二台服务器上安装了hello2程序。通过应用注册页面，将两个应用注册后，在APP Center主页面分别选择两个应用执行，图0是选择执行hello1程序的运行效果。

图0

图0是选择执行hello2程序的运行效果。

图0

从运行结果可见，只要用户使用正确的信息注册了应用，在使用应用的过程中，只需要非常简单的点击操作，配合适当输入的参数信息，就可以成功实现云应用的运行，用户体验在原先程序的基础上有了较大提高。至此，我们就完成了AppScale上的APP Center应用的开发任务。

接下来，我们在AWS云平台上开发大数据和人工智能相关的几个典型应用，并将这些应用集成到本节开发的APP Center中，实现所有应用的统一管理和运维。

## 第2节大数据实验

spark是一个快速且通用的集群计算平台。快速指扩充了MapReduce计算模型，基于内存的计算。通用是指Spark的设计容纳了其他分布式系统的功能，如批处理、迭代计算，交互查询和流处理等。Spark是开放的，虽然使用scala语言编写，但内置多种语言的API。Hadoop应用场景大多用于离线处理，对时效性要求不高，与之相比，Spark应用在时效性要求高、机器学习等领域。Spark组件紧密集成，节省了各个组件使用时的部署，测试等时间，同时集成的组件也得到相应的优化。每个组件都各善其职，都有特定的作用，应用于不同的场景，组成了一个生态系统。

在AWS平台上订阅预装Spark环境的过程和上一节中订阅AppScale环境的过程基本一致，只是在选择虚拟机模板这一步不同，需要搜索Spark字样的模板，如图0所示。

搜索框输入spark后回车确认

![](images/spark_aws/003.png)

寻找合适镜像，此处选择的是`Apache Spark Powered by Code Creator`

![](images/spark_aws/004.png)

选择后继续

![](images/spark_aws/005.png)

根据需要选择所需的硬件资源，若不需要额外配置，此处可直接点击`审核和启动`完成实例的创建

![](images/spark_aws/006.png)

后续过程和订阅AppScale环境一样。订阅成功后，使用ssh登录系统

```

ssh -i spark.pem ubuntu@ec2-54-169-37-43.ap-southeast-1.compute.amazonaws.com

```

尝试运行，出现`scala`命令提示符，表示spark环境安装成功。如图0所示。

图0

在scale提示符后输入spark相关指令，即可进行大数据相关处理任务。

下面通过几个实验，展示spark的几个典型应用，并将这些应用发布到上一节中完成的应用中心，更好的体现出云应用的使用模式。

### 实验1 词频统计

**实验目的**

统计文本文件中英文单词的出现频率，并按照从高到低的顺序排列。

**实验要求**

使用HDFS存放数据，使用Spark分析词频，实验结果发布到AppScale应用中心。

**实验步骤**

首先准备一个测试用文本文件`input.txt`，文件内容为任意一篇英文文章。

本实验仅要求统计英文单词的出现频率，无需考虑标点符号和其他特殊符号，所以，先对文本文件`input.txt`进行预处理，剔除文本中的标点符号和其他特殊符号，程序名为`pre-process.py`，代码如下：

```

#!/usr/bin/python

with open('input.txt', 'r') as f:

text = f.read().lower()

for i in '!@#$%^&*()_+-;:`~\'"<>=./?,':

text = text.replace(i, '')

with open('output.txt', 'w') as f:

f.write(text)

```

其中，第一行 `#!/usr/bin/python` 表示这是一个python程序。 `text = f.read().lower()`一行将文件中的所有大写字母转换为小写字母，方便后续统计。`for ... text = text.replace(i, '')` 通过一个for循环，依次把每一个特殊字符用空格替换掉，这是因为题目需求里面并不要求统计特殊字符的出现次数。最后一条语句`f.write(text)`把处理完后的文本输出到文件`output.txt`中。

接下来将预处理后的文件`output.txt`上传到分布式文件系统HDFS中，方便后续Spark分布式处理。此步骤不是必须的，Spark也可以处理普通文件系统上的文件。但是，如果把文件存放到分布式文件系统HDFS中，再调用Spark进行处理，能够更好的发挥出Spark分布式并行分析数据的功能。

```

hdfs dfs -mkdir /spark

hdfs dfs -put output.txt /spark

```

接下来，执行spark-shell，进入交互式命令行界面，如下所示：

```

spark-shell

scala>

```

将`output.txt`文件内容读入变量data，指令如下：

```

val data = sc.textFile("/spark/output.txt")

```

查看data内容：

```

scala> data.collect

res1: Array[String] = Array(the story of an hour, by kate chopin, this story was first published in 1894 as the dream of an hour before being republished under this title in 1895 we encourage students ...

```

从输出内容可见，data是一个数组变量，存放了`output.txt`文件的全部文本，用换行符分割文本，每行文本用一个数组元素存放。为了进行词频统计，我们需要按空格分割文本，让每个数组元素存放一个单词，指令如下：

```

val words = data.flatMap(line => line.split(" "))

```

查看分割后的结果：

```

scala> words.collect

res2: Array[String] = Array(the, story, of, an, hour, by, kate, chopin, this, story, was, first, published, in, 1894, as, the, dream, of, an, hour, before, being, republished, under, this, title, in, 1895, we, encourage, students ...

```

为统计每个单词的出现次数，先把每个单词扩展为一个二元组：第一元为单词本身，第二元为单词出现的次数（初始时为1），代码如下：

```

val wordsmap = words.map(word => (word, 1))

```

扩展后的结果为：

```

scala> wordsmap.collect

res3: Array[(String, Int)] = Array((the,1), (story,1), (of,1), (an,1), (hour,1), (by,1), (kate,1), (chopin,1), (this,1), (story,1), (was,1), (first,1), (published,1), (in,1), (1894,1), (as,1), (the,1), (dream,1), (of,1), (an,1) ...

```

接下来把全部第一元相同的二元组合并为一个新二元组，新二元组第二元记录全部原二元组第二元之和，代码如下：

```

val wordscount = wordsmap.reduceByKey((x,y) => x+y)

```

合并后的结果如下：

```

scala> wordscount.collect

res5: Array[(String, Int)] = Array((controversy,1), (chose,1), (someone,2), (under,2), (08,1), (reading,1), (its,3), (guide,2), (opening,1), (gripsack,1), (bright,1), (warmed,1), (filled,1), (have,6), (carried,1), (we,1) ...

```

对结果排序，按第二元数值从大到小排列，代码如下：

```

val wordssort = wordscount.sortBy(-_._2)

```

排序后的结果如下：

```

scala> wordssort.collect

res6: Array[(String, Int)] = Array((the,67), (her,41), (of,39), (and,37), (she,34), (a,33), (was,30), (to,29), (in,26), ("",24), (that,24), (it,15), (with,13), (had,13), (as,12), (story,12), (there,10), (not,8), (would,8) ...

```

将结果存放到`/spark/result`中：

```

wordssort.saveAsTextFile("/spark/result")

```

最后执行`:quit`指令，退出`spark shell`，回到Linux命令行中。执行以下指令，把结果从分布式文件系统HDFS中下载到服务器本地：

```

hdfs dfs -get /spark/result

```

执行完该指令后，当前目录下会生成一个result目录，目录中包含了程序的执行结果：

```

# ls

input.txt output.txt pre-process.py result

# ls result/

part-00000 part-00001 _SUCCESS

```

查看第一个结果文件的前10行，即可看到按照词频排序，出现次数最多的前10个单词：

```

# head result/part-00000

(the,4394)

(to,4200)

(of,3696)

(and,3637)

(her,2249)

(i,2105)

(a,1959)

(in,1892)

(was,1860)

(she,1725)

```

至此，词频统计的任务就完成了。接下来，我们把这个应用集成到第1节实验3中实现的应用中心。目前，词频统计的任务是在Spark-shell交互式环境下完成的，需要人工输入和执行每一条指令，获得一条指令的结果后，才能输入下一条指令并执行。这样的模式无法直接结合到应用中心。所以，首先需要对程序的执行过程进行改造，变成一条指令即可完成全部任务的形式。

首先新建一个文本文件count.scala，把所有spark指令集成到一起，如下所示：

```

val data = sc.textFile("/spark/output.txt")

val words = data.flatMap(line => line.split(" "))

val wordsmap = words.map(word => (word, 1))

val wordscount = wordsmap.reduceByKey((x,y) => x+y)

val wordssort = wordscount.sortBy(-_._2)

wordssort.saveAsTextFile("/spark/result")

```

再新建一个脚本文件count.sh，该文件会读取count.scala文件内容并逐一执行其中的spark指令，代码如下：

```

#!/bin/bash

./pre-process.py

hdfs dfs -rm -f -r /spark/output.txt

hdfs dfs -put output.txt /spark

hdfs dfs -rm -f -r /spark/result

spark-shell -i < count.scala >/dev/null 2>&1

rm -fr result/

hdfs dfs -get /spark/result

head result/part-00000

```

其中第一行`#!/bin/bash`表示这是一个bash脚本文件。`spark-shell -i < count.scala >/dev/null 2>&1`一行启动`spark-shell，并从count.scala文件中逐一读取spark指令并执行，执行过程中忽略掉所有的的屏幕提示信息。这行代码之前的内容是把待处理文件上传到HDFS中，之后的代码是将处理结果从HDFS中下载到本地，并显示出来。先尝试在Linux命令行中执行该脚本文件，检查结果是否正确，如下所示：

```

# ./count.sh

...

(the,4394)

(to,4200)

(of,3696)

(and,3637)

(her,2249)

(i,2105)

(a,1959)

(in,1892)

(was,1860)

(she,1725)

```

可见，程序执行结果和之前在Spark-shell交互式环境下逐一执行每一条指令的最终结果是一样的。接下来，我们就可以把这个程序集成到AppScale应用中心了。通过应用注册页面，将词频统计应用注册到应用中心，注册页面如图0所示。

图0 注册应用

注册成功后，在APP Center主页面选择词频统计应用并执行，图0是应用的执行效果。

图0

至此，我们就完成了词频统计应用的开发任务。

### 实验2. 事故分析

**实验目的**

根据某城市某年的交通事故数据（[数据集地址](HTTP://data.gov.uk/dataset/cb7ae6f0-4be6-4935-9277-47e5ce24a11f/road-safety-data)），分析天气情况、时间、地点或其它因素与交通事故之间的关系。

**实验要求**

使用Spark分析，实验结果发布到AppScale应用中心。

**实验步骤**

从公共数据集中选择一个作为本实验分析的对象（例如，选择2018年的交通事故数据集），将该数据集文件`dftRoadSafetyData_Accidents_2018.csv`下载到本地。该数据集包含了事故编号、事发地点、时间，事故严重程度，事故车辆类型、道路类型、路面干湿情况、天气情况、光线情况等30多个字段。本实验先分析天气情况与交通事故之间的关系。

首先在HDFS中新建accident目录，把交通事故数据集文件`dftRoadSafetyData_Accidents_2018.csv`和天气情况文件`weather.csv`上传到HDFS：

```

hdfs dfs -mkdir /accident

hdfs dfs -put dftRoadSafetyData_Accidents_2018.csv /accident

hdfs dfs -put weather.csv /accident

```

其中，交通事故数据集文件`dftRoadSafetyData_Accidents_2018.csv`格式如下：

<br>图0. 交通事故数据集文件格式<br></div>

天气情况文件`weather.csv`格式如下：

<br>图0. 天气情况文件格式<br></div>

把交通事故数据读入变量accidentsDF：

```

scala> val accidentsDF = spark.read.option("header", "true").option("inferSchema", "true").csv("/accident/dftRoadSafetyData_Accidents_2018.csv")

```

其中，`option("header", "true")`表示文件第一行为表头信息，`option("inferSchema", "true")`表示用spark内置的检测机制判断读入数据数据的类型。接下来把天气信息读入变量weatherDF：

```

scala> val weatherDF = spark.read.option("header", "true").csv("/accident/weather.csv")

```

从全部交通事故记录中，提取严重程度大于1的记录：

```

scala> val results1 = accidentsDF.filter(accidentsDF.col("Accident_Severity") > 1)

```

在交通事故文件中，`Accident_Index`一列的前4个字符表示事故发生的年份，截取这个数据，起一个别名`year`，选取`year`和`Weather_Conditions`两个字段：

```

scala> val results2=results1.select(col("Accident_Index").substr(0, 4).as("year"), col("Weather_Conditions"))

```

使用`year`和`Weather_Condition`这两个字段进行分组，统计组内数据的个数。

```

scala> val results3=results2.groupBy("year", "Weather_Conditions").count()

```

为了使显示结果更加直观，把用数字代表的`Weather_Condition`转换成文字：

```

scala> val results4=results3.join(weatherDF, weatherDF.col("code") === accidentsDF.col("Weather_Conditions"))

```

这里使用了join操作和天气情况文件`weather.csv`完成这个任务。接下来选取：年份、天气情况和事故数量这几个字段，并按照事故数量从多到少的顺序排序：

```

scala> val results5=results4.select(col("year"), col("label").as("weather"), col("count")).orderBy(col("count").desc)

```

最后把结果显示出来：

```

scala> results5.show

+----+----------------------------+-----+

|year|weather |count|

+----+----------------------------+-----+

|2018|Fine no high winds |97851|

|2018|Raining no high winds |12628|

|2018|Unknown |3639 |

|2018|Other |2582 |

|2018|Raining + high winds |1247 |

|2018|Fine + high winds |1107 |

|2018|Snowing no high winds |1063 |

|2018|Fog or mist |428 |

|2018|Snowing + high winds |400 |

|2018|Data missing or out of range|19 |

+----+----------------------------+-----+

```

从分析结果可知，交通事故发生次数最多的天气是"Fine no high winds"，即良好且无大风的天气。这种天气发生事故的次数远高于雨、雪、雾的天气，这似乎我们“恶劣天气更易发生事故”的常识相违背，其实不然，因为这里我们仅仅统计了事故的次数，并没有统计当年未发生交通事故的事件。通常来说，天气较好的时候人们更愿意出门，从而导致事故次数变多，这并不能说明天气良好更容易发生事故。另一方面，天气良好的情况下，人们的警惕性相较于特殊天气会降低，这也可能是导致天气良好的情况下事故较多的原因。

用同样的方法，可以分析时间（这里的时间指的是每周的星期几）与交通事故之间的关系。只需要将上述步骤中读取天气状况的地方，改为读取时间即可，具体修改之处如下。

将记录时间信息的文件`week.csv`上传到HDFS：

```

hdfs dfs -put week.csv /accident

```

该文件格式如下：

<br>图0. 时间信息文件格式<br></div>

把时间信息读入变量weekDF：

```

scala> val weekDF = spark.read.option("header", "true").csv("/accident/week.csv")

```

选取`year`和`Day_of_Week`两个字段：

```

scala> val results2=results1.select(col("Accident_Index").substr(0, 4).as("year"), col("Day_of_Week"))

```

使用`year`和`Day_of_Week`这两个字段进行分组，统计组内数据的个数。

```

scala> val results3=results2.groupBy("year", "Day_of_Week").count()

```

使用join操作和时间信息文件`week.csv`，把用数字代表的`Day_of_Week`转换成文字：

```

scala> val results4=results3.join(weekDF, weekDF.col("code") === accidentsDF.col("Day_of_Week"))

```

选取：年份、时间和事故数量这几个字段，并按照事故数量从多到少的顺序排序：

```

scala> val results5=results4.select(col("year"), col("label").as("week"), col("count")).orderBy(col("count").desc)

```

最后把结果显示出来：

```

scala> results5.show

+----+---------+-----+

|year| week|count|

+----+---------+-----+

|2018| saturday|19776|

|2018| friday|18438|

|2018| thursday|18196|

|2018|wednesday|17734|

|2018| tuesday|17489|

|2018| sunday|15786|

|2018| monday|13545|

+----+---------+-----+

```

从分析结果可知，在一周中交通事故发生次数最多的时间是周期六，发生次数最少的时间是星期一，这跟“周末驾车出游人数较多，交通事故发生概率更大”的常识是相符的。

上述过程分别分析了天气情况和时间与交通事故之间的关系，且两个任务都是在Spark-shell交互式环境下完成的。现在将两个任务合并到一个应用中，通过参数控制执行哪一个任务，以方便后续将应用发布到AppScale应用中心。新建脚本文件`accident.sh`，文件内容如下：

```

#!/bin/bash

if [[ $1 = 'weather' ]]; then

spark-shell -i < weather.scala

elif [[ $1 = 'week' ]]; then

spark-shell -i < week.scala

else

echo "please input parameter: weather | week"

```

其中，`weather.scala`和`week.scala `分别是将上述两个任务中，在Spark-shell交互式环境下执行的所有spark指令集成到一起形成的文本文件。脚本文件`accident.sh`根据第一个输入参数`$1`是`weather`还是`week`，分别读取`weather.scala`和`week.scala `文件的内容，并逐一执行其中的spark指令。当输入参数为`weather`时，程序执行效果如下：

```

# ./accident.sh weather

...

+----+----------------------------+-----+

|year|weather |count|

+----+----------------------------+-----+

|2018|Fine no high winds |97851|

|2018|Raining no high winds |12628|

|2018|Unknown |3639 |

|2018|Other |2582 |

|2018|Raining + high winds |1247 |

|2018|Fine + high winds |1107 |

|2018|Snowing no high winds |1063 |

|2018|Fog or mist |428 |

|2018|Snowing + high winds |400 |

|2018|Data missing or out of range|19 |

+----+----------------------------+-----+

```

当输入参数为`week`时，程序执行效果如下：

```

# ./accident.sh week

...

+----+---------+-----+

|year| week|count|

+----+---------+-----+

|2018| saturday|19776|

|2018| friday|18438|

|2018| thursday|18196|

|2018|wednesday|17734|

|2018| tuesday|17489|

|2018| sunday|15786|

|2018| monday|13545|

+----+---------+-----+

```

最后，我们把这个程序集成到AppScale应用中心了。通过应用注册页面，将事故分析应用注册到应用中心，注册页面如图0所示。

图0 注册应用

注册成功后，在APP Center主页面选择事故分析应用，输入适当的参数并执行，图0是应用的执行效果。

图0

至此，我们就完成了事故分析应用的开发任务。

### 实验3. 电影推荐

**实验目的**

根据某电影论坛的公开数据集（[数据集地址](https://grouplens.org/datasets/movielens/)），分析用户对电影的评分情况，并向用户推荐可能喜欢的电影。

**实验要求**

使用Spark分析，实验结果发布到AppScale应用中心。

**实验步骤**

首先在HDFS中新建movie目录，把电影文件`movies.csv`和评分文件`ratings.csv`上传到HDFS：

```

hdfs dfs -mkdir /movie

hdfs dfs -put movies.csv /movie

hdfs dfs -put ratings.csv /movie

```

其中，电影文件`movies.csv`格式如下：

<br>图0. 电影文件格式<br></div>

评分文件`ratings.csv`格式如下：

<br>图0. 评分文件格式<br></div>

读入数据，文件格式是csv，可选参数里指明该CSV文件有头部，并且将类型推断开关打开。类型推断会将`userId`,`movieId`推断为integer类型，`rating`推断为double类型，`timestamp`推断为integer类型，这符合我们对数据的要求，不需要额外操作。可使用`printSchema()`函数查看各字段信息：

```

scala> val ratingsDF = spark.read.option("header", "true").option("inferSchema", "true").csv("/movie/ratings.csv")

scala> ratingsDF.printSchema()

root

|-- userId: integer (nullable = true)

|-- movieId: integer (nullable = true)

|-- rating: double (nullable = true)

|-- timestamp: integer (nullable = true)

```

对于常用的数据分析任务，Spark提供了很多功能完善的库，我们可以直接使用这些库来简化自己工作。本实验会用到Random和ALS两个库，先导入这两个库：

```

scala> import scala.util.Random

scala> import org.apache.spark.ml.recommendation.ALS

```

新建一个ALS库提供的模型`als`，并对这个模型进行初始化：

```

scala> :paste

val als = new ALS()

.setSeed(Random.nextLong())

.setImplicitPrefs(true)

.setRank(10)

.setRegParam(0.01)

.setAlpha(1.0)

.setMaxIter(5)

.setUserCol("userId")

.setItemCol("movieId")

.setRatingCol("rating")

.setPredictionCol("prediction")

```

使用刚读入的评分文件`ratings.csv`训练`als`模型：

```

scala> val model = als.fit(ratingsDF)

```

读入电影数据文件`movies.csv`，字段有`movieId`、`title`、`genres`，分别表示电影ID、电影名称、电影类型，其中类型由"|"分隔。

接下来尝试选择`userId=274`的用户来看推荐效果。先统计该用户的观看过电影的类型：

```

scala> val userId = 274

scala> val watchedMoviesId = ratingsDF.filter($"userId" === userId).select("movieId").as[Int].collect()

scala> :paste

moviesDF.filter($"movieId" isin (watchedMoviesId:_*))

.withColumn("genres", explode(split($"genres", "\\|")))

.select("genres")

.groupBy("genres")

.count()

.orderBy($"count".desc)

.show()

+-----------+-----+

| genres|count|

+-----------+-----+

| Comedy| 550|

| Action| 454|

| Thriller| 452|

| Drama| 407|

| Adventure| 295|

| Horror| 266|

| Sci-Fi| 232|

| Crime| 224|

| Fantasy| 175|

| Children| 158|

| Romance| 132|

| Mystery| 109|

| Animation| 99|

| Musical| 49|

| War| 34|

| IMAX| 34|

| Western| 26|

|Documentary| 13|

| Film-Noir| 7|

+-----------+-----+

```

从上面可以看出该用户看过的最多的类型有：喜剧、动作、惊悚和戏剧类型。现在我们给该用户推荐10部电影。我们使用ALS类的`recommendForUserSubset`方法为指定的用户集推荐电影，此处用户集`userSubset`中只有一个用户，其ID为274。

```

scala> val userSubset = ratingsDF.select(als.getUserCol).filter($"userId" === userId).distinct().limit(1)

scala> val recommendations = model.recommendForUserSubset(userSubset, 10)

```

查看推荐结果：

```

scala> recommendations.show(truncate=false)

...|userId|recommendations...

...|274 |[[2804, 1.3512053], [1387, 1.3346307], [5679, 1.3175706], [54503, 1.3016284], [8641, 1.2686287], [3275, 1.2642331], [51662, 1.2634813], [8783, 1.262716], [1258, 1.2550267], [8874, 1.240392]]...

```

为了更直观显示推荐结果，可继续对结果进行处理。首先获取推荐电影的ID：

```

scala> val recsMoviesId = recommendations.withColumn("recommendations", explode($"recommendations")).select("recommendations.movieId").as[Int].collect()

```

在结果中显示出推荐电影的名称和类型：

```

scala> moviesDF.filter($"movieId" isin (recsMoviesId:_*)).show(10, false)

+-------+--------------------------------------------+---------------------------+

|movieId|title |genres |

+-------+--------------------------------------------+---------------------------+

|1258 |Shining, The (1980) |Horror |

|8641 |Anchorman: The Legend of Ron Burgundy (2004)|Comedy |

|51662 |300 (2007) |Action|Fantasy|War|IMAX |

|54503 |Superbad (2007) |Comedy |

+-------+--------------------------------------------+---------------------------+

```

实验结果较好的反映了用户爱好的电影名称和类型。但是该推荐结果仍存在一定的缺陷：推荐了部分用户已经看过的电影。主要原因在于，该用户评价了1000多部电影，评分高于2.5分的就有1035部，而本次训练使用的数据集只有约9000部电影，数据集太小，导致训练的结果不够理想，如果使用更大的数据集训练模型，推荐效果会有所提高。

上述实验过程是在Spark-shell交互式环境下进行的，并且只尝试了一个用户`userId=274`的情况。现在将全部过程集成到一个脚本文件`movie.sh`中，通过参数指定为哪一个用户推荐电影，文件内容如下：

```

#!/bin/bash

if [[ $1 -lt 1 || $1 -gt 610 ]];

then

echo "Usage: ./movie_recommend.sh userid(1~610)"

exit -1

userId=$1

sed -i '10c var userId = '${userId}'' movie.scala

spark-shell < movie.scala

```

其中，`movie.scala`是将上述Spark-shell交互式环境下执行的所有spark指令集成到一起形成的文本文件。脚本文件`movie.sh`从第一个输入参数`$1`是中获取用户ID，然后使用`sed`指令将`movie.scala`文件中的`userId`替换为该用户ID，再逐一执行其中的spark指令。当输入参数为`274`时，程序执行效果如下：

```

# ./movie.sh 274

...

+-------+--------------------------------------------+---------------------------+

|movieId|title |genres |

+-------+--------------------------------------------+---------------------------+

|1258 |Shining, The (1980) |Horror |

|8641 |Anchorman: The Legend of Ron Burgundy (2004)|Comedy |

|51662 |300 (2007) |Action|Fantasy|War|IMAX |

|54503 |Superbad (2007) |Comedy |

+-------+--------------------------------------------+---------------------------+

```

最后，我们把这个程序集成到AppScale应用中心了。通过应用注册页面，将电影推荐应用注册到应用中心，注册页面如图0所示。

图0 注册应用

注册成功后，在APP Center主页面选择电影推荐应用，在参数框中输入用户ID并执行，图0是应用的执行效果。

图0

至此，我们就完成了电影推荐应用的开发任务。

## 第3节机器学习实验

Deep Learning（深度学习）是机器学习领域中一个新的研究方向，它被引入机器学习使其更接近于最初的目标——人工智能。Deep Learning是学习样本数据的内在规律和表示层次，这些学习过程中获得的信息对诸如文字，图像和声音等数据的解释有很大的帮助。它的最终目标是让机器能够像人一样具有分析学习能力，能够识别文字、图像和声音等数据。Deep Learning是一个复杂的机器学习算法，在语音和图像识别方面取得的效果，远远超过先前相关技术。Deep Learning在搜索技术，数据挖掘，机器学习，机器翻译，自然语言处理，多媒体学习，语音，推荐和个性化技术，以及其他相关领域都取得了很多成果。Deep Learning使机器模仿视听和思考等人类的活动，解决了很多复杂的模式识别难题，使得人工智能相关技术取得了很大进步。Deep Learning以神经网络算法为起源，以模型结构深度的增加而发展，随着大数据和计算能力的提高产生了一系列新的算法。

两种最常见的深度学习框架——TensorFlow和Pytorch。TensorFlow 是一个开源的、基于 Python 的机器学习框架，它由 Google 开发，并在图形分类、音频处理、推荐系统和自然语言处理等场景下有着丰富的应用，是目前最热门的机器学习框架。除了 Python，TensorFlow 也提供了 C/C++、Java、Go、R 等其它编程语言的接口。PyTorch 是一个 Python 优先的深度学习框架，能够在强大的 GPU 加速基础上实现张量和动态神经网络。

在AWS平台上订阅预装Deep Learning环境的过程和上一节中订阅AppScale环境的过程基本一致，只是在选择虚拟机模板这一步不同，需要选择Deep Learning字样的模板，如图0所示。

![1570257334506](./images/deep_learning/1570257334506.png)

2. 选择实例类型

![1570254062790](./images/deep_learning/1570254062790.png)

3. 配置实例详细信息

![1570254116509](./images/deep_learning/1570254116509.png)

4. 添加存储

![1570254143952](./images/deep_learning/1570254143952.png)

5.添加标签

![1570254166798](./images/deep_learning/1570254166798.png)

6.配置安全组

![1570254212216](./images/deep_learning/1570254212216.png)

7.审核

![1570254246020](./images/deep_learning/1570254246020.png)

8.选择密钥对并下载

![1570254354134](./images/deep_learning/1570254354134.png)

9. ssh连接

修改pem文件权限

```

chmod 400 tensorflow_new.pem

```

查看IP

![1570261365084](./images/deep_learning/1570261365084.png)

ssh连接

```

ssh -i tensorflow_new.pem ubuntu@ec2-13-250-104-250.ap-southeast-1.compute.amazonaws.com

```

10.测试

查看所有的环境

```

conda info -e

```

![1570261484191](./images/deep_learning/1570261484191.png)

加载环境

```

source activate tensorflow_p36

```

测试代码

![1570261859897](./images/deep_learning/1570261859897.png)

显示以上信息表示深度学习环境搭建成功。

### 实验1. 数字识别

**数据集介绍**

手写数字识别可以说是深度学习中的入门级案例，我们将借助于机器的力量完成视觉方面的内容，我们的手写数字识别用的数据集是mnist数据集，他的结构如下：

| 文件 | 内容 |

| ------------------------------------------------------------ | ------------------------------------------------ |

| [`train-images-idx3-ubyte.gz`](http://yann.lecun.com/exdb/mnist/train-images-idx3-ubyte.gz) | 训练集图片 - 55000 张训练图片, 5000 张验证图片 |

| [`train-labels-idx1-ubyte.gz`](http://yann.lecun.com/exdb/mnist/train-labels-idx1-ubyte.gz) | 训练集图片对应的数字标签 |

| [`t10k-images-idx3-ubyte.gz`](http://yann.lecun.com/exdb/mnist/t10k-images-idx3-ubyte.gz) | 测试集图片 - 10000 张图片 |

| [`t10k-labels-idx1-ubyte.gz`](http://yann.lecun.com/exdb/mnist/t10k-labels-idx1-ubyte.gz) | 测试集图片对应的数字标签 |

这些都是压缩包的格式，当我们解压缩之后——借助于read_data_sets（）函数就可以完成——得到的结构如下：

| 数据集 | 使用用途 |

| ---------------------- | --------------------------- |

| `data_sets.train` | 55000 组图片和标签, 训练集 |

| `data_sets.validation` | 5000 组图片和标签, 验证集 |

| `data_sets.test` | 10000 组图片和标签, 测试集 |

**环境准备**

- windows中matplotlib库的下载

1. 点击pycharm中File中的Settings

![1570177375735](./images/deep_learning/1570177375735.png)

2. 点击Project interpreter

![1570177440739](./images/deep_learning/1570177440739.png)

3. 点击右边的“+”，输入matplotlib，安装即可

![1570177729256](./images/deep_learning/1570177729256.png)

- 测试：

? 打开Anaconda Prompt，输入"activate tensorflow"启动tensorflow环境，后输入"python"进入python环境

? 输入测试代码

![测试截图](./images/deep_learning/测试截图.png)

输出内容，测试完毕，安装成功

- Mnist数据集下载

网址如下：[点击进入下载](http://yann.lecun.com/exdb/mnist/)

![1570175579942](./images/deep_learning/1570175579942.png)

**使用方式**

+ 项目的框架构成：

![1570175803224](./images/deep_learning/1570175803224.png)

1. 第一个部分是mnist数据集，将之前下载好的数据集copy到mnist目录下

2. 第二个部分是tensorflow图的权重和偏置值的保存，底下为固定的保存的格式

3. 第三个部分为我们的主程序，即为训练手写数字识别的主函数

+ 执行的方式

直接运行main.py即可运行

+ 程序的输入

1. 第一种输入格式直接调用test测试集中的images，我们也可以使用matplotlib库中自带的imshow（）函数进行显示

例如：

![1570176839399](./images/deep_learning/1570176839399.png)

2. 第二种输入格式是我们自己去制作图片，但是这种制作图片的方式并不是任意的，需要时黑底白字，即背景颜色全部是黑色，中间的数字为白色才可以。

这一种输入方式我们可以借助于画图工具完成，但需要注意一个地方：

我们的程序接受的图片的大小为28 \* 28，所以我们需要修正图片的大小为28 \* 28, 这一步我们同样可以借助于cv2——python中的一种图像处理库——的reshape（）函数。

+ 程序的输出

程序的输出很明显就是去对我们输入的图片进行解析并说明这幅图片是什么结果，为了方便和源图片进行比对，我们可以将输出的结果和源图片放在一起进行输出。

这一步我们可以借助于matplotlib库中的text（）函数完成。

+ 运行的示例

![](./images/deep_learning/运行结果.png)

>对于结果的解释：

>1. 这是一个3*3的图，每一个图形都是一张数字，表示这是我们输入进去的图像的信息

>2. 每个图像的正下方是一个输出的预估的结果，即通过我们训练后由预估器给出的结果

>3. 可以发现大部分的预测的结果都是很准确的，只有极少一部分的结果是错误的，我们实现的手写数字识别的准确率是很高的

**结语**

这虽然只是tensorflow的一个入门级程序，但是我们已经可以从中发现一些深度学习的玄妙之处，我们可以借助机器的力量实现一些我们生物才可以完成的事情，希望读者可以好好领会其中的思想，并且加以运用。

### 实验2. 智能写诗

**实验内容**

利用pytorch网络，训练一个智能写诗机器人

**环境准备**

> 上文中已经介绍了如何部署基于TensorFlow的深度学习环境，在这个基础上，我们再来介绍下如何部署pytorch框架

**当前环境说明：**Ubuntu 16.04 + Python3.6

- 安装pytorch

在pytorch官网(https://pytorch.org/)，根据自己的环境进行选择：

在shell中运行:`pip3 install torch torchvision`

等待安装完毕即可。

- numpy等基本的Python包一般系统都已自带，在遇到缺少相关库文件时，直接利用`pip3`包管理器进行安装即可。

**运行写诗机器人**

- 克隆项目仓库 `git clone git@github.com:braveryCHR/LSTM_poem.git`

项目结构：

- data.py -- 预处理数据

- config.py -- 配置网络脚本

- checkpoints -- 训练结果

- main.py -- 训练脚本

- test.py -- 测试脚本

- 下载数据集 http://pytorch-1252820389.cosbj.myqcloud.com/tang_199.pth

- 根据情况修改网络`config.py`

```Python

class Config(object):

num_layers = 3 # LSTM层数

data_path = 'data/' # 诗歌的文本文件存放路径

pickle_path = 'tang.npz' # 预处理好的二进制文件

author = None # 只学习某位作者的诗歌

constrain = None # 长度限制

category = 'poet.tang' # 类别，唐诗还是宋诗歌(poet.song)

lr = 1e-3

weight_decay = 1e-4

use_gpu = False

epoch = 50

batch_size = 16

maxlen = 125 # 超过这个长度的之后字被丢弃，小于这个长度的在前面补空格

plot_every = 200 # 每20个batch 可视化一次

# use_env = True # 是否使用visodm

env = 'poetry' # visdom env

max_gen_len = 48 # 生成诗歌最长长度

debug_file = '/tmp/debugp'

model_path = "./checkpoints/tang_new.pth" # 预训练模型路径

prefix_words = '岁岁年年花相似,年年岁岁人不同' # 不是诗歌的组成部分，用来控制生成诗歌的意境

start_words = '闲云潭影日悠悠' # 诗歌开始

acrostic = False # 是否是藏头诗

```

- 训练模型 `python3 main.py`

- 查看训练效果 `python3 test.py`

![image-20191004190604219](./images/deep_learning/image-20191004190604219.png)

无标题文章

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读