微信搜索：“二十同学” 公众号，欢迎关注一条不一样的成长之路

Flink 为流处理和批处理分别提供了 DataStream API 和 DataSet API。正是这种高层的抽象和 flunent API 极大地便利了用户编写大数据应用。

不过很多初学者在看到官方 Streaming 文档中那一大坨的转换时，常常会蒙了圈，文档中那些只言片语也很难讲清它们之间的关系。

本文将介绍几种关键的数据流类型，它们之间是如何通过转换关联起来的。

下图展示了 Flink 中目前支持的主要几种流的类型，以及它们之间的转换关系。

DataStream

DataStream 是 Flink 流处理 API 中最核心的数据结构。它代表了一个运行在多个分区上的并行流。一个 DataStream 可以从 StreamExecutionEnvironment 通过env.addSource(SourceFunction) 获得。

DataStream 上的转换操作都是逐条的，比如 map()，flatMap()，filter()。DataStream 也可以执行 rebalance（再平衡，用来减轻数据倾斜）和 broadcaseted（广播）等分区转换。

val stream: DataStream[MyType] = env.addSource(new FlinkKafkaConsumer08[String](...))
val str1: DataStream[(String, MyType)] = stream.flatMap { ... }
val str2: DataStream[(String, MyType)] = stream.rebalance()
val str3: DataStream[AnotherType] = stream.map { ... }

上述 DataStream 上的转换在运行时会转换成如下的执行图：

如上图的执行图所示，DataStream 各个算子会并行运行，算子之间是数据流分区。

如 Source 的第一个并行实例（S1）和 flatMap() 的第一个并行实例（m1）之间就是一个数据流分区，而在 flatMap() 和 map() 之间由于加了 rebalance()，它们之间的数据流分区就有3个子分区（m1的数据流向3个map()实例）。

这与 Apache Kafka 是很类似的，把流想象成 Kafka Topic，而一个流分区就表示一个 Topic Partition，流的目标并行算子实例就是 Kafka Consumers。

KeyedStream

KeyedStream用来表示根据指定的key进行分组的数据流。

一个KeyedStream可以通过调用DataStream.keyBy()来获得，而在KeyedStream上进行任何transformation都将转变回DataStream。

在实现中，KeyedStream是把key的信息写入到了transformation中。每条记录只能访问所属key的状态，其上的聚合函数可以方便地操作和保存对应key的状态。

WindowedStream & AllWindowedStream

WindowedStream代表了根据key分组，并且基于WindowAssigner切分窗口的数据流。

所以WindowedStream都是从KeyedStream衍生而来的，而在WindowedStream上进行任何transformation也都将转变回DataStream。

val stream: DataStream[MyType] = ...
val windowed: WindowedDataStream[MyType] = stream
.keyBy("userId")
.window(TumblingEventTimeWindows.of(Time.seconds(5))) // Last 5 seconds of data
val result: DataStream[ResultType] = windowed.reduce(myReducer)

上述 WindowedStream 的样例代码在运行时会转换成如下的执行图：

Flink 的窗口实现中会将到达的数据缓存在对应的窗口buffer中（一个数据可能会对应多个窗口）。

当到达窗口发送的条件时（由Trigger控制），Flink 会对整个窗口中的数据进行处理。Flink 在聚合类窗口有一定的优化，即不会保存窗口中的所有值，而是每到一个元素执行一次聚合函数，最终只保存一份数据即可。

在key分组的流上进行窗口切分是比较常用的场景，也能够很好地并行化（不同的key上的窗口聚合可以分配到不同的task去处理）。不过有时候我们也需要在普通流上进行窗口的操作，这就是 AllWindowedStream。AllWindowedStream是直接在DataStream上进行windowAll(...)操作。

AllWindowedStream 的实现是基于 WindowedStream 的（Flink 1.1.x 开始）。Flink 不推荐使用AllWindowedStream，因为在普通流上进行窗口操作，就势必需要将所有分区的流都汇集到单个的Task中，而这个单个的Task很显然就会成为整个Job的瓶颈。

JoinedStreams & CoGroupedStreams

双流 Join 也是一个非常常见的应用场景。深入源码你可以发现，JoinedStreams 和 CoGroupedStreams 的代码实现有80%是一模一样的，JoinedStreams 在底层又调用了 CoGroupedStreams 来实现 Join 功能。除了名字不一样，一开始很难将它们区分开来，而且为什么要提供两个功能类似的接口呢？？

实际上这两者还是很点区别的。首先 co-group 侧重的是group，是对同一个key上的两组集合进行操作，而 join 侧重的是pair，是对同一个key上的每对元素进行操作。co-group 比 join 更通用一些，因为 join 只是 co-group 的一个特例，所以 join 是可以基于 co-group 来实现的（当然有优化的空间）。而在 co-group 之外又提供了 join 接口是因为用户更熟悉 join（源于数据库吧），而且能够跟 DataSet API 保持一致，降低用户的学习成本。

JoinedStreams 和 CoGroupedStreams 是基于 Window 上实现的，所以 CoGroupedStreams 最终又调用了 WindowedStream 来实现。

val firstInput: DataStream[MyType] = ...
val secondInput: DataStream[AnotherType] = ...

val result: DataStream[(MyType, AnotherType)] = firstInput.join(secondInput)
.where("userId").equalTo("id")
.window(TumblingEventTimeWindows.of(Time.seconds(3)))
.apply (new JoinFunction () {...})

上述 JoinedStreams 的样例代码在运行时会转换成如下的执行图：

双流上的数据在同一个key的会被分别分配到同一个window窗口的左右两个篮子里，当window结束的时候，会对左右篮子进行笛卡尔积从而得到每一对pair，对每一对pair应用 JoinFunction。

不过目前（Flink 1.1.x）JoinedStreams 只是简单地实现了流上的join操作而已，距离真正的生产使用还是有些距离。

因为目前 join 窗口的双流数据都是被缓存在内存中的，也就是说如果某个key上的窗口数据太多就会导致 JVM OOM（然而数据倾斜是常态）。

双流join的难点也正是在这里，这也是社区后面对 join 操作的优化方向，例如可以借鉴Flink在批处理join中的优化方案，也可以用ManagedMemory来管理窗口中的数据，并当数据超过阈值时能spill到硬盘。

ConnectedStreams

在 DataStream 上有一个 union 的转换 dataStream.union(otherStream1, otherStream2, ...)，用来合并多个流，新的流会包含所有流中的数据。union 有一个限制，就是所有合并的流的类型必须是一致的。ConnectedStreams 提供了和 union 类似的功能，用来连接两个流，但是与 union 转换有以下几个区别：

1. ConnectedStreams 只能连接两个流，而 union 可以连接多于两个流。
2. ConnectedStreams 连接的两个流类型可以不一致，而 union 连接的流的类型必须一致。
3. ConnectedStreams 会对两个流的数据应用不同的处理方法，并且双流之间可以共享状态。这在第一个流的输入会影响第二个流时, 会非常有用。

如下 ConnectedStreams 的样例，连接 input 和 other 流，并在input流上应用map1方法，在other上应用map2方法，双流可以共享状态（比如计数）。

val input: DataStream[MyType] = ...
val other: DataStream[AnotherType] = ...

val connected: ConnectedStreams[MyType, AnotherType] = input.connect(other)

val result: DataStream[ResultType] = 
connected.map(new CoMapFunction[MyType, AnotherType, ResultType]() {
override def map1(value: MyType): ResultType = { ... }
override def map2(value: AnotherType): ResultType = { ... }
})

当并行度为2时，其执行图如下所示：

总结

本文介绍通过不同数据流类型的转换图来解释每一种数据流的含义、转换关系。后面的文章会深入讲解 Window 机制的实现，双流 Join 的实现等。

首先：我先先来看看实现的效果：

然后我再说说具体实现的步骤吧：（别急文件后面我会把百度网盘链接发出来的，除了小案例外还有jsMind的文件，案例，目前的说明文档）

 第一：引入jsMind的css.js文件

第二：页面添加一个容器：

第三：也是最重要的js部分了（后面会有整个页面的代码）

好了，知道贴图大家肯定不得劲，我也是，网上找的资料不能拷贝真的很痛苦，下面是代码：

<!DOCTYPE html>
<html>
<head>
	<title>jsmindd</title>
</head>
<link rel="stylesheet" href="css/jsmind.css">
<script type="text/javascript" src="js/jquery-2.1.1.min.js"></script>
<script type="text/javascript" src="js/jsmind.js"></script>
<script type="text/javascript" src="js/jsmind.draggable.js"></script>
<body>
	 <div id="jsmind_container" style="width: 1000px;height: 1000px;"></div>
</body>
</html>
<script>
		var mind = {
            "meta":{
                "name":"jsMind remote",
                "author":"hizzgdev@163.com",
                "version":"0.2"
            },
            "format":"node_tree",
            "data":{"id":"root","topic":"jsMind","children":[
                {"id":"easy","topic":"Easy","direction":"left","children":[
                    {"id":"easy1","topic":"Easy to show"},
                    {"id":"easy2","topic":"Easy to edit"},
                    {"id":"easy3","topic":"Easy to store"},
                    {"id":"easy4","topic":"Easy to embed"},
                    {"id":"other3","background-image":"ant.png", "width": "100", "height": "100"}
                ]},
                {"id":"open","topic":"Open Source","direction":"right","children":[
                    {"id":"open1","topic":"on GitHub", "background-color":"#eee", "foreground-color":"blue"},
                    {"id":"open2","topic":"BSD License"}
                ]},
                {"id":"powerful","topic":"Powerful","direction":"right","children":[
                    {"id":"powerful1","topic":"Base on Javascript"},
                    {"id":"powerful2","topic":"Base on HTML5"},
                    {"id":"powerful3","topic":"Depends on you"}
                ]},
                {"id":"other","topic":"test node","direction":"left","children":[
                    {"id":"other1","topic":"I'm from local variable"},
                    {"id":"other2","topic":"I can do everything"}
                ]}
            ]}
        }
		// jsMind的选项，更多参数参见jsMind的文档
		// https://github.com/hizzgdev/jsmind/blob/master/docs/zh/index.md
		var options = {
		    container:'jsmind_container',//容器的ID
		    editable:false,				// 是否启用编辑
		    theme:'primary'				//主题

		    	//options的属性
		    	//container : '',         // [必选] 容器的ID
			   // editable : false,       // 是否启用编辑
			   // theme : null,           // 主题
			   // mode :'full',           // 显示模式========full - 子节点动态分布在根节点两侧 [默认值] side - 子节点只分布在根节点右侧
			   // support_html : true,    // 是否支持节点里的HTML元素
			   // view:{
			   //     hmargin:100,        // 思维导图距容器外框的最小水平距离
			   //     vmargin:50,         // 思维导图距容器外框的最小垂直距离
			   //     line_width:2,       // 思维导图线条的粗细
			   //     line_color:'#555'   // 思维导图线条的颜色
			   // },
			   // layout:{
			   //     hspace:30,          // 节点之间的水平间距
			   //     vspace:20,          // 节点之间的垂直间距
			   //     pspace:13           // 节点与连接线之间的水平间距（用于容纳节点收缩/展开控制器）
			   // },
			   // shortcut:{
			   //     enable:true,        // 是否启用快捷键
			   //     handles:{},         // 命名的快捷键事件处理器
			   //     mapping:{           // 快捷键映射
			   //         addchild   : 45,    // <Insert>
			   //         addbrother : 13,    // <Enter>
			   //         editnode   : 113,   // <F2>
			   //         delnode    : 46,    // <Delete>
			   //         toggle     : 32,    // <Space>
			   //         left       : 37,    // <Left>
			   //         up         : 38,    // <Up>
			   //         right      : 39,    // <Right>
			   //         down       : 40,    // <Down>
			   //     }
			   // },
		}
		var jm = new jsMind(options);


		jm.show(mind);
		//jm.disable_edit();//禁止编制
		jm.expand_all();//展开全部节点
		jm.add_node(parent_node, nodeid, topic, data);//添加节点


		console.log(jm.get_root());
	
</script>

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

好了，这个思维导图就完成了，下面首先放百度网盘的链接。里面会有jsmind的css,jsmind的js，还有一个可以拖拽的js文件，

第二份是jsMind的文件，里面有官方案例，还有操作文档那些，最后我会把握了解的一些关于jsmind的内容分享一下

百度网盘地址：

1、里面是css,js

https://pan.baidu.com/s/1cH_dkFFZZni9PRz2sUU1bg      提取码：vycc

2、官方案例，文件，文档

已失效 ：https://pan.baidu.com/s/18sLQhD3hC-gTXYOVPkwa6Q     提取码： nrpv      

最新：https://pan.baidu.com/s/1KA3RN7FS1_5300xQmfdyXA     提取码： 28bc

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

好了，关于这个思维导图的相关文件都在这里了，下面我在说说对jsMind的一些个人理解吧，因为我也百度了好多资料，发现对jsmind的一些相关资料是相当少的，可能大家用的也不多吧，在通过研究官方文档以后有了一些新的看法，

1、jsMind 支持三种数据格式，分别是树对象格式、表对象格式、freemind格式----这里资料摘取自官方文档

• 树对象格式 默认格式，节点之间是包含关系，便于程序进行处理，适合与 MongoDB 及其它文档型数据库进行数据交互；
• 表对象格式 节点之间是并列关系，使用 parentid 标识上下级关系，适合与关系型数据库进行数据交互；
• freemind格式 使用 freemind 的 xml 格式，适合与 freemind 进行数据交互。    

  // 三种格式
  		A. 树对象格式示例
  		var mind = {
  				    /* 元数据，定义思维导图的名称、作者、版本等信息 */
  				    "meta":{
  				        "name":"jsMind-demo-tree",
  				        "author":"hizzgdev@163.com",
  				        "version":"0.2"
  				    },
  				    /* 数据格式声明 */
  				    "format":"node_tree",
  				    /* 数据内容 */
  				    "data":{"id":"root","topic":"jsMind","children":[
  				        {"id":"easy","topic":"Easy","direction":"left","expanded":false,"children":[
  				            {"id":"easy1","topic":"Easy to show"},
  				            {"id":"easy2","topic":"Easy to edit"},
  				            {"id":"easy3","topic":"Easy to store"},
  				            {"id":"easy4","topic":"Easy to embed"}
  				        ]},
  				        {"id":"open","topic":"Open Source","direction":"right","expanded":true,"children":[
  				            {"id":"open1","topic":"on GitHub"},
  				            {"id":"open2","topic":"BSD License"}
  				        ]},
  				        {"id":"powerful","topic":"Powerful","direction":"right","children":[
  				            {"id":"powerful1","topic":"Base on Javascript"},
  				            {"id":"powerful2","topic":"Base on HTML5"},
  				            {"id":"powerful3","topic":"Depends on you"}
  				        ]},
  				        {"id":"other","topic":"test node","direction":"left","children":[
  				            {"id":"other1","topic":"I'm from local variable"},
  				            {"id":"other2","topic":"I can do everything"}
  				        ]}
  				    ]}
  				};
  		B. 表对象格式示例
  		var mind = {
  			    /* 元数据，定义思维导图的名称、作者、版本等信息 */
  			    "meta":{
  			        "name":"example",
  			        "author":"hizzgdev@163.com",
  			        "version":"0.2"
  			    },
  			    /* 数据格式声明 */
  			    "format":"node_array",
  			    /* 数据内容 */
  			    "data":[
  			        {"id":"root", "isroot":true, "topic":"jsMind"},
  
  			        {"id":"easy", "parentid":"root", "topic":"Easy", "direction":"left"},
  			        {"id":"easy1", "parentid":"easy", "topic":"Easy to show"},
  			        {"id":"easy2", "parentid":"easy", "topic":"Easy to edit"},
  			        {"id":"easy3", "parentid":"easy", "topic":"Easy to store"},
  			        {"id":"easy4", "parentid":"easy", "topic":"Easy to embed"},
  
  			        {"id":"open", "parentid":"root", "topic":"Open Source", "expanded":false, "direction":"right"},
  			        {"id":"open1", "parentid":"open", "topic":"on GitHub"},
  			        {"id":"open2", "parentid":"open", "topic":"BSD License"},
  
  			        {"id":"powerful", "parentid":"root", "topic":"Powerful", "direction":"right"},
  			        {"id":"powerful1", "parentid":"powerful", "topic":"Base on Javascript"},
  			        {"id":"powerful2", "parentid":"powerful", "topic":"Base on HTML5"},
  			        {"id":"powerful3", "parentid":"powerful", "topic":"Depends on you"},
  			    ]
  			};	
  		C. freemind格式示例
  		var mind = {
  			    /* 元数据，定义思维导图的名称、作者、版本等信息 */
  			    "meta":{
  			        "name":"example",
  			        "author":"hizzgdev@163.com",
  			        "version":"0.2"
  			    },
  			    /* 数据格式声明 */
  			    "format":"freemind",
  			    /* 数据内容 */
  			    "data":"<map version=\"1.0.1\"> <node ID=\"root\" TEXT=\"jsMind\" > <node ID=\"easy\" POSITION=\"left\" TEXT=\"Easy\" > <node ID=\"easy1\" TEXT=\"Easy to show\" /> <node ID=\"easy2\" TEXT=\"Easy to edit\" /> <node ID=\"easy3\" TEXT=\"Easy to store\" /> <node ID=\"easy4\" TEXT=\"Easy to embed\" /> </node> <node ID=\"open\" POSITION=\"right\" TEXT=\"Open Source\" > <node ID=\"open1\" TEXT=\"on GitHub\" /> <node ID=\"open2\" TEXT=\"BSD License\" /> </node> <node ID=\"powerful\" POSITION=\"right\" TEXT=\"Powerful\" > <node ID=\"powerful1\" TEXT=\"Base on Javascript\" /> <node ID=\"powerful2\" TEXT=\"Base on HTML5\" /> <node ID=\"powerful3\" TEXT=\"Depends on you\" /> </node> <node ID=\"other\" POSITION=\"left\" TEXT=\"test node\" > <node ID=\"other1\" TEXT=\"I'm from local variable\" /> <node ID=\"other2\" TEXT=\"I can do everything\" /> </node> </node> </map>"
  			};	

  注：
  		除 freemind 格式外，其余两种格式的基本数据结构如下：
  			    {
  			        "id":"open",           // [必选] ID, 所有节点的ID不应有重复，否则ID重复的结节将被忽略
  			        "topic":"Open Source", // [必选] 节点上显示的内容
  			        "direction":"right",   // [可选] 节点的方向，此数据仅在第一层节点上有效，目前仅支持 left 和 right 两种，默认为 right
  			        "expanded":true,       // [可选] 该节点是否是展开状态，默认为 true
  			    }	

  好了，最主要的还是说这个数据格式的不同，三种，树状和表格我们可能用的多一点，第三种很少用了，最底下我会把网址分享一下，感兴趣的朋友可以深入研究下，这里面的文档更详细：

•                                    https://github.com/hizzgdev/jsmind/blob/master/docs/zh/1.usage.md

Android应用双开是某些特殊人群的需要，目前已知的双开方案个人总结为3种：

1.反编译apk，然后修改uid等相关信息让系统弄认为apk有不同，然后重新安装。这个方法是简单粗暴的，不过目前主流的app估计都会预防这种修改方法的。对于手机厂商来说，拥有整个系统的代码权限，不会用这么低级的手段。此种方法我个人也无深入研究。

2.插件化，在启动代理apk后动态加载需要双开的apk，此种方法是双开实现中最复杂最有技术含量的实现方式。插件化开源的有360 发布的DroidPlugin，未开源最有名气的就是lbe的MultiDroid，据我所知MultiDroid也是诸多手机厂商使用的较多的方案。此种方案的好处是无需修改系统代码，问题隔离的完善，对已有系统没有任何影响。关于插件化的原理见http://www.jianshu.com/u/e347b97e2f0c，weishu的多篇文章讲解的很透彻了。

3.多用户，小米使用的是这种实现方案，也是我个人想到双开第一个冒出的想法就是多用户。用户系统是Android系统支持的完整的一套东西，基于此开发就相当于站在巨人的肩膀上。基于账户系统的双开本来以为有一定的难度，要改不少地方，最近看了不少UserManagerService的文章和相关源码，最后发现一行代码不改就能看到双开的效果。

见原生代码的demo

development/samples/browseable/BasicManagedProfile

此demo的本意是展示android5.0的新特性，设备管理managed profiles 。源码相关就是DevicePolicyManager，它可以限制app的行为，例如无法使用摄像头，浏览器只能打开固定几个网页等等。基本原理是新建了一个拥有FLAG_MANAGED_PROFILE flag的user，并且创建想要控制的app到该user，然后控制app行为。该user的app和原有的app都会显示在launcher上面，这个是市面主流的双开app需要的所要的效果是一样一样的.......

不仅桌面同时显示，按history键显示的历史堆栈会显示所有打开的app，两个user的同app也都会显示。

原生的demo只显示了system app的使用，如果是三方app需要修改几行代码。

development/samples/browseable/BasicManagedProfile/src/com/example/android/basicmanagedprofile/BasicManagedProfileFragment.java

private void setAppEnabled(String packageName, boolean enabled) {
                    
//                    devicePolicyManager.enableSystemApp(
//                            BasicDeviceAdminReceiver.getComponentName(activity), packageName);
                   
                      int userId = UserHandle.myUserId();
                      packageManager.installExistingPackageAsUser(packageName, userId);

    }

注释的代码是原有代码，后面的是我针对三方app写的代码。注意AndroidManifest中要加相应的权限

    <uses-permission android:name="android.permission.INSTALL_PACKAGES"/>
    <uses-permission android:name="android.permission.DELETE_PACKAGES"/>
    <uses-permission android:name="android.permission.INTERACT_ACROSS_USERS" />
    <uses-permission android:name="android.permission.INTERACT_ACROSS_USERS_FULL" />

基本功能ok了，其余的双开需求就是苦力活了。

再次证明了对系统代码熟悉的话，实现功能可能异常的简单。