本节首先分析传统流处理系统架构存在的问题，然后介绍 Spark Streaming 的系统架构及其工作原理和优势。
流处理架构的分布式流处理管道执行方式是，首先用数据采集系统接收来自数据源的流数据，然后在集群上并行处理数据，最后将处理结果存放至下游系统。
为了处理这些数据，传统的流处理系统被设计为连续算子模型，其工作方式如图 1 所示。
系统包含一系列的工作结点，每组结点上运行一至多个连续算子。对于流数据，每个连续算子（ContinuousOperator）一次处理一条记录，并且将记录传输给管道中别的算子，源算子（SourceOperator）从采集系统接收数据，接着沉算子（SinkOperator）输出到下游系统。
连续算子是一种较为简单、自然的模型。然而，在大数据时代，随着数据规模的不断扩大，以及越来越复杂的实时分析，这个传统的架构面临着严峻的挑战。
TRUNCATE [TABLE] 表名
其中，TABLE 关键字可省略。
mysql> CREATE TABLE `tb_student_course` (
-> `id` int(4) NOT NULL AUTO_INCREMENT,
-> `name` varchar(25) NOT NULL,
-> PRIMARY KEY (`id`)
-> );
Query OK, 0 rows affected (0.04 sec)
mysql> INSERT INTO tb_student_course(name) VALUES ('Java'),('MySQL'),('Python');
Query OK, 3 rows affected (0.05 sec)
Records: 3 Duplicates: 0 Warnings: 0
mysql> SELECT * FROM tb_student_course;
+----+--------+
| id | name |
+----+--------+
| 1 | Java |
| 2 | MySQL |
| 3 | Python |
+----+--------+
3 rows in set (0.00 sec)
使用 TRUNCATE 语句清空 tb_student_course 表中的记录，SQL 语句和运行结果如下：
mysql> TRUNCATE TABLE tb_student_course; Query OK, 0 rows affected (0.04 sec) mysql> SELECT * FROM tb_student_course; Empty set (0.00 sec)
class django.db,model.ForeignKey(to,on_delete,**options)
#一个A类实例对象关联多个B类实例对象
class A(model.Model):
....
class B(model.Model):
属性 = models.ForeignKey(多对一中"一"的模型类, ...)
from django.db import models
#新建出版社表
class PubName(models.Model):
pubname=models.CharField('名称',max_length=255,unique=True)
#更改书籍信息表
class Book(models.Model):
title=models.CharField(max_length=30,unique=True, verbose_name='书名')
price=models.DecimalField(max_digits=7,decimal_places=2,verbose_name='定价')
#添加默认价格
def default_price(self):
return '￥30'
#零售价格
retail_price=models.DecimalField(max_digits=7,decimal_places=2,verbose_name='零售价',default=default_price)
pub=models.ForeignKey(to=PubName,on_delete=models.CASCADE ,null=True) #创建Foreign外键关联pub,以pub_id关联
def __str__(self)
return "title:%s pub:%s price:%s" % (self.title, self.pub, self.price)
此处需要注意每次更改完 models 都需要进行数据库迁移操作，依次执行以下命令即可：
python manager.py makemigrations
python manager.py migrate
#创建PubName实例化对象pub1并插入书籍信息 pub1=PubName.objects.create(pubname="清华出版社") Book.objects.create(title="Python",price="59.00",retail_price="59.00",pub=pub1) Book.objects.create(title="Redis",price="25.00",retail_price="25.00",pub=pub1) Book.objects.create(title="Java",price="45.00",retail_price="45.00",pub=pub1) #创建PubName实例化对象pub2并插入书籍信息 pub2=PubName.objects.create(pubname="c出版") Book.objects.create(title="Django",price="65.00",retail_price="65.00",pub=pub2) Book.objects.create(title="Flask",price="45.00",retail_price="45.00",pub=pub2) Book.objects.create(title="Tornado",price="35.00",retail_price="35.00",pub=pub2)访问 MySQL 数据库分别查询 index_book、index_pubname 数据表（如下所示），index_pubname 数据表的 id 字段作为唯一值关联多个书籍信息，ForeignKey 外键关联键自动在 index_book 表中生成 pub_Id 字段并作为关联字段。此时 index_pubname 作为主表而 index_book 是子表，主表的 id 是子表的外键，两者之间存在外键约束 CASCADE。
mysql> select * from index_book; +----+---------+--------+-------+--------------+ | id | title | pub_id | price | retail_price | +----+---------+--------+-------+--------------+ | 1 | Python | 1 | 59.00 | 59.00 | | 2 | Redis | 1 | 25.00 | 25.00 | | 3 | Java | 1 | 45.00 | 45.00 | | 4 | Django | 2 | 65.00 | 65.00 | | 5 | Flask | 2 | 45.00 | 45.00 | | 6 | Tornado | 2 | 35.00 | 35.00 | +----+---------+--------+-------+--------------+ 6 rows in set (0.01 sec) mysql> select * from index_pubname; +----+-----------------+ | id | pubname | +----+-----------------+ | 1 | c出版 | | 2 | 清华出版社 | +----+-----------------+ 6 rows in set (0.00 sec)
class A(model.Model):
...
class B(model.Model):
属性 = models.OneToOneField(A)
#新建一对一关用户信息表拓展表,添加完成后执行数据库迁移同步操作
class ExtendUserinfo(models.Model):
user=models.OneToOneField(to=UserInfo,on_delete=models.CASCADE)
signature=models.CharField(max_length=255,verbose_name='用户签名',help_text='自建签名')
nickname=models.CharField(max_length=255,verbose_name='昵称',help_text='自建昵称')
使用 Django shell 创建数据，如下所示：from index.models import UserInfo,ExtendUserinfo username=UserInfo.objects.create(username="xiaoming",password="******") username=UserInfo.objects.create(username="xiaohong",password="*******",gender="F") #创建一对一表关联 ExtendUserinfo.objects.create(user=username,signature="good good study,day day up",nickname="XH")
mysql> select * from index_userinfo; +----+----------+----------+--------+ | id | username | password | gender | +----+----------+----------+--------+ | 1 | xiaoming | ****** | M | | 2 | xiaohong | ******* | F | +----+----------+----------+--------+ 2 rows in set (0.00 sec) mysql> select * from index_extenduserinfo; +----+----------------------------+----------+---------+ | id | signature | nickname | user_id | +----+----------------------------+----------+---------+ | 1 | good good study,day day up | XH | 2 | +----+----------------------------+----------+---------+ 1 row in set (0.00 sec)
多对多关系也是比较常见的，比如一个作者可以写很多本书，一本书也可以由很多作者一起完成，那么这时候 Author 和 Book 之间就是多对多的关系。 Django 通过中间表的方式来实现 Model 之间的多对多的关系，这和 MySQL 中实现方式是一致的。这个中间表我们可以自己提供，也可以使用 Django 默认生成的中间表。
class django.db.models.ManyToManyFiled(to,**options)
它只有一个必填的参数即 to，与其他两个关联词在一样，用来指定与当前的 Model 关联的 Model。
class Author(models.Model):
...
class Book(models.Model):
...
authors = models.ManyToManyField(Author)
books=models.ManyToManyField(to="Book") #创建多对多映射关系然后再执行数据库迁移命令，我们可以执行以下命令可以查看 Django 执行 sql 语句：
python manage.py sqlmigrate index 0007_author_books
sql 语句如下所示：CREATE TABLE `index_author_books` (`id` integer AUTO_INCREMENT NOT NULL PRIMARY KEY, `author_id` integer NOT NULL, `book_id` integer NOT NULL); ALTER TABLE `index_author_books` ADD CONSTRAINT `index_author_books_author_id_2bfd143c_fk_index_author_id` FOREIGN KEY (`author_id`) REFERENCES `index_author` (`id`); ALTER TABLE `index_author_books` ADD CONSTRAINT `index_author_books_book_id_1c280bc9_fk_index_book_id` FOREIGN KEY (`book_id`) REFERENCES `index_book` (`id`); ALTER TABLE `index_author_books` ADD CONSTRAINT `index_author_books_author_id_book_id_b0dd3503_uniq` UNIQUE (`author_id`, `book_id`);由 sql 语句可以看出，Django 默认隐式的创建了 index_author_books 表（Django 的命名规范），即维护关联关系的中间表。这个表有三个字段分别是主键 id，与index_author 表关联的 author_id，以及 index_book 表关联的 book_id。同时为这两个关联 id 创建了外键约束（FORGIEN KEY）,最后还为 index_author_books 表创建了唯一性约束 author_id 和 book_id。
author1=Author.objects.create(name="Luncy",email="123456@qq.com") author2=Author.objects.create(name="Tom",email="456789@163.com")因为书籍信息之前已经准备完毕，所以下面我们开始创建多对多映射关系，我们在 Django shell 进行如下操作：
author1.books.add(Book.objects.get(id="1")) author1.books.add(Book.objects.get(id="2")) author1.books.add(Book.objects.get(id="3")) author2.books.add(Book.objects.get(id="1")) author2.books.add(Book.objects.get(id="4")) author2.books.add(Book.objects.get(id="5")) author2.books.add(Book.objects.get(id="3")) author2.books.add(Book.objects.get(id="6")) author1.books.add(Book.objects.get(id="6"))多对多关系在中间表插入数据需要使用 add() 方法，books 是对应的多对多字段。通过以上代码就完成多对多关系的创建，最后在 MySQL 中查看多对多相关联的三张数据表，如下所示：
#书籍信息表（index_book） +----+---------+--------+-------+--------------+ | id | title | pub_id | price | retail_price | +----+---------+--------+-------+--------------+ | 1 | Python | 8 | 59.00 | 59.00 | | 2 | Redis | 8 | 25.00 | 25.00 | | 3 | Java | 8 | 45.00 | 45.00 | | 4 | Django | 9 | 65.00 | 65.00 | | 5 | Flask | 9 | 45.00 | 45.00 | | 6 | Tornado | 9 | 35.00 | 35.00 | +----+---------+--------+-------+--------------+ #作家信息表（index_author） +----+-------+----------------+ | id | name | email | +----+-------+----------------+ | 1 | Luncy | 123456@qq.com | | 2 | Tom | 456789@163.com | +----+-------+----------------+ #中间表（index_author_books） +----+-----------+---------+ | id | author_id | book_id | +----+-----------+---------+ | 4 | 1 | 1 | | 7 | 1 | 3 | | 5 | 1 | 4 | | 6 | 1 | 5 | | 8 | 1 | 6 | | 1 | 2 | 1 | | 2 | 2 | 2 | | 3 | 2 | 3 | | 9 | 2 | 6 | +----+-----------+---------+本节用了较长的篇幅给大家讲解了 Django 中数据表的关联关系，它和 MySQL 的思想是一致的，只是 Django 提供了自己的一套方法，所以我们也要学会使用它，在后续章节我们将基于此节的内容介绍 Django QuerySet API 即与数据库接口相关的表查询、更新、删除操作。
举个例子（实例一）：显然此方式违背了“声明位于 .h 文件，定义（实现）位于 .cpp 文件”的规律。在 C++ 多文件编程中，还有 2 种特殊情况是违背此规律的，分别是类的定义和内联函数的定义，通常情况下它们也都定义在 .h 文件中。
//demo.h
#ifndef _DEMO_H
#define _DEMO_H
const int num = 10;
#endif
//main.cpp
#include <iostream>
#include"demo.h"
int main() {
std::cout << num << std::endl;
return 0;
}
项目执行结果为：
10
可以看到，使用此方式定义的 const 常量，只需引入其头文件（比如这里的 demo.h），就可以使用它。注意，将 const 常量定义在 .h 文件中，为了避免头文件被重复引入，推荐使用 #ifndef/#define/#endif 结构。除此之外，还有 2 中方式可以避免头文件被重复引入，读者可阅读《C++防止头文件被重复引入的3种方法》一文做详细了解。
//demo.h
#ifndef _DEMO_H
#define _DEMO_H
extern const int num; //声明 const 常量
#endif
//demo.cpp
#include "demo.h" //一定要引入该头文件
const int num =10; //定义 .h 文件中声明的 num 常量
//main.cpp
#include <iostream>
#include "demo.h"
int main() {
std::cout << num << std::endl;
return 0;
}
项目执行结果为：
10
前面讲过，C++ const 关键字会限定变量的可见范围为当前文件，即无法在其它文件中使用该常量。而 extern 关键字会 const 限定可见范围的功能，它可以使 const 常量的可见范围恢复至整个项目。
//demo.cpp
extern const int num =10;
//main.cpp
#include <iostream>
extern const int num;
int main() {
std::cout << num << std::endl;
return 0;
}
项目执行结果为：
10
显然相比实例二，此项目中省略了 demo.h 头文件的创建，一定程序上提高了项目的编译效率。本节介绍了 3 种在多文件编程中使用 const 常量的方法，相比后 2 种借助 extern 修饰 const 常量的方式，第一种方式更简单、更常用，推荐读者使用。
@。将其放置在一个 PHP 表达式之前，该表达式可能产生的任何错误信息都将被忽略掉。 @ 的话，错误处理函数将返回 0。@ 运算符只对表达式有效。简单来说就是，如果能从某处得到值，就能在它前面加上 @ 运算符。例如可以在变量、函数、include 调用、常量等等之前使用 @ 运算符，但不能把它放在函数或类的定义之前，也不能用于条件结构例如 if 和 foreach 等语句前。@ 运算符对于可以导致程序终止的严重错误也是有效的，这意味着如果在某个不存在或者敲错了字母的函数调用前用了 @ 来抑制错误信息，那么程序将没有任何提示的死在那里。
<?php
$link = @mysqli_connect("127.0.0.1", "my_user", "my_password", "my_db") or die('数据库连接失败！');
?>
运行结果如下：
数据库连接失败！
error_reporting([int $level])
其中参数 $level 为设置错误级别，具体可以使用的值可以在《PHP错误级别》一节中查看。如果将 $level 设置为 0，将关闭所有 PHP 错误报告；如果设置为 -1，将返回所有的错误报告。
<?php
error_reporting(0);
$link = mysqli_connect("127.0.0.1", "my_user", "my_password", "my_db") or die('数据库连接失败！');
?>
; This directive controls whether or not and where PHP will output errors,
; notices and warnings too. Error output is very useful during development, but
; it could be very dangerous in production environments. Depending on the code
; which is triggering the error, sensitive information could potentially leak
; out of your application such as database usernames and passwords or worse.
; For production environments, we recommend logging errors rather than
; sending them to STDOUT.
; Possible Values:
; Off = Do not display any errors
; stderr = Display errors to STDERR (affects only CGI/CLI binaries!)
; On or stdout = Display errors to STDOUT
; Default Value: On
; Development Value: On
; Production Value: Off
; http://php.net/display-errors
display_errors = Off
Django shell 是 Django 提供一种交互环境，类似于 Python shell，开发人员可直接通过 shell 环境 与 Django 项目进行交互，它一种很实用的工具。
python manage.py shell
然后创建 Model 实例，使用 save() 方法保存数据，如下所示：from index.models import Book #导入相关model类 #创建Book实例化对象 book=Book(title="Python",public="a",price="59.00",retail_price="59.00") book.save()# 调用save方法进行保存 book=Book(title="Flask",public="b",price="39.00",retail_price="39.00") book.save()# 调用save方法进行保存 book=Book(title="Django",public="b",price="40.00",retail_price="40.00") book.save()#最后调用 save 方法进行保存。通过上述代码，我们就在数据库中添加了三条有关书籍的记录。
mysql> select * from index_book; +----+--------+--------+-------+--------------+ | id | title | public | price | retail_price | +----+--------+--------+-------+--------------+ | 1 | Python | a | 59.00 | 59.00 | | 2 | Flask | b | 39.00 | 39.00 | | 3 | Django | b | 40.00 | 40.00 | +----+--------+--------+-------+--------------+ 3 rows in set (0.00 sec)避坑提示，若在实例化数据对象的过程中，Django shell 出现以这样的报错信息：
InternalError: (1366, "Incorrect string value: '\\xE8\\xAF\\xAD\\xE8\\xA8\\x80...' for column 'public' at row 1")
说明我们添加的 public 字段的信息超过可字段本身的最大长度限制，即 manx_length=50。解决方法是将字段 max_length 值变大然后需要重新执行数据库迁移的操作，或者将你输入的信息缩短从而符合其规定的长度。book.title="JavaScript" book.save()
In [1]: Book.objects.create(title="Java",public="a",price="30.00",retail_price="30.00") Out[1]: <Book: title:Java public:a price:30.00> #执行create方法存在返回值然后继续插入如下数据：
from index.models import Book In [2]: Book.objects.create(title="MySQL",public="c",price="35.00",retail_price="35.00") Out[2]: <Book: title:MySQL public:c price:35.00> In [3]: Book.objects.create(title="C#",public="a",price="45.00",retail_price="40.00") Out[3]: <Book: title:C# public:a price:45.00> In [4]: Book.objects.create(title="Redis",public="c",price="25.00",retail_price="25.00") Out[4]: <Book: title:Redis public:c price:25.00>这种创建 Model 实例对象的过程比较简单，create 执行之后就在数据库中插入了一条记录,这也是我们推荐使用的方法。最后访问数据库可以看到最新插入的数据记录：
mysql> select * from index_book; +----+--------+--------+-------+--------------+ | id | title | public | price | retail_price | +----+--------+--------+-------+--------------+ | 1 | Python | a | 59.00 | 59.00 | | 2 | Flask | b | 39.00 | 39.00 | | 3 | Django | b | 40.00 | 40.00 | | 4 | Java | a | 30.00 | 30.00 | | 5 | MySQL | c | 35.00 | 35.00 | | 6 | C# | a | 45.00 | 40.00 | | 7 | Redis | c | 25.00 | 25.00 | +----+--------+--------+-------+--------------+本节我们完成了向数据表插入数据记录的操作，不管是在 shell 环境下，还是项目的代码中都可以使用以上两种方法实现创建 Model 实例。后续章节会将涉及 Django 表的数据查询，我们还会用到本节内容。
DoesNotExist异常。而当这两个方法的查询条件都能够匹配多条数据记录时，都会抛出MultipleObjectsReturned异常。提示：这两个方法都只能返回一条数据。
#使用 Django shell Author.objects.create(name="Tom",email="456789@163.com") #添加 Tom 此时数据表有两个Tom Author.objects.get(name="Tom") #查询 name="Tom",就会报错 #报错信息如下 MultipleObjectsReturned: get() returned more than one Author -- it returned 2! #查询不存在数据 Author.objects.get(id=4) #报错信息如下： DoesNotExist：Traceback (most recent call last)
Author.objects.get_or_create(name="Xiaolong")#先查询是否存在若不存在则新建该实例对象 (<Author: 作家：Xiaolong>, True) #返回值是一个元组有两个元素元组中的第一个元素代表实例对象，第二个元素是布尔值，标识返回的实例对象是否是新创建的，其中 True 代表新建实例对象，False 代表原有实例对象。
查询结果为单实例的类似方法还有 first、last 等方法，用法与语法格式别无二致，在这里就不加赘述了。
MyModel.objects.raw('sql语句')
def BookName(request):
books=Book.objects.raw("select * from index_book") #书写sql语句
return render(request,"index/allbook.html",locals())
{% for book in books %}
<p>{{book.title}}</p>
{% endfor %}
最后配置路由，通过访问 127.0.0.1:8000/index/allbook/，可以得到所有书籍的 title。
authors=Author.objects.raw("select id from index_author where name= %s",['Tom'])
...: for author in authors:
...: print('%s:%s'%(author.id,author.name))
2:Tom #返回结果
raw 方法可以接受一个列表或字典类型的参数，并将 SQL 语句中的占位符替换，最终完成查询。
from django.db import connection
用创建 cursor 类的构造函数创建 cursor 对象，再使用 cursor 对象执行 SQL 语句。为确保能够在出现异常时释放 cursor 游标，通常使用 with 语句进行创建操，如下所示：
from django.db import connection
with connection.cursor() as cur:
cur.execute('执行SQL语句')
使用示例如下：
from django.db import connection
with connection.cursor() as cur:
#调用游标对象的execute方法，更新author的名字
cur.execute('update index_author set name="Jack" where id=3;')
with connection.cursor() as cur:
# 删除id为3的一条author记录
cur.execute('delete from index_author where id=3;')
虽然 Django 提供了 raw SQL的查询方法以及 cursor 游标对象的使用，但是它们很少的被使用，应为这些方法都需要考虑不同数据库的不同特性。Django 官网建议我们尽量使用 ORM 模块来完成相关操作。