你好，四月

2018-04-01

与谁诉愁肠越来越无助，千般娇媚眼前浮，我也想颓废，成为心中期待的自己
好像很孤单，一个人看那静止的时间，
漫长又短暂无趣又无味，多么难忘的滋味。

我欲欲何为，竟成了无为，用时间苦水教我什么是人生
没明白却明白了很久，不懂得却懂得了很多
只是都给了文字没给现实，怎么感觉自己很无知，连简单的道理都不明白
是不是太失败，不敢想也不敢做，
还是想多了容易乱，容易不快了。

上了眉头无人诉，平淡成了奢望的玩笑，没有追求成了碌碌无为
任岁月流逝光阴不复，我本该有所为却又无所为，难道就这样过完一生
我不甘心中还想挑战，偶尔又萌生了胆怯。
不知何时喜欢上了安稳，就想留在父母身边，现实一次又一次的告诉我你要有安稳的条件，不然就成了懒惰。

more >>

展开全文 >>

如何解决数据倾斜？

2016-08-23

1.数据倾斜调优

调优概述
有的时候，我们可能会遇到大数据计算中一个最棘手的问题——数据倾斜，此时Spark作业的性能会比期望差很多。数据倾斜调优，就是使用各种技术方案解决不同类型的数据倾斜问题，以保证Spark作业的性能。

数据倾斜发生时的现象

绝大多数task执行得都非常快，但个别task执行极慢。比如，总共有1000个task，997个task都在1分钟之内执行完了，但是剩余两三个task却要一两个小时。这种情况很常见。

原本能够正常执行的Spark作业，某天突然报出OOM（内存溢出）异常，观察异常栈，是我们写的业务代码造成的。这种情况比较少见。

数据倾斜发生的原因

数据倾斜的原因很简单：在进行shuffle的时候，必须将各个节点上相同的key拉取到某个节点上的一个task来进行处理，比如按照key进行聚合或join等操作。此时如果某个key对应的数据量特别大的话，就会发生数据倾斜。比如大部分key对应10条数据，但是个别key却对应了100万条数据，那么大部分task可能就只会分配到10条数据，然后1秒钟就运行完了；但是个别task可能分配到了100万数据，要运行一两个小时。因此，整个Spark作业的运行进度是由运行时间最长的那个task决定的。

因此出现数据倾斜的时候，Spark作业看起来会运行得非常缓慢，甚至可能因为某个task处理的数据量过大导致内存溢出。

下图就是一个很清晰的例子：hello这个key，在三个节点上对应了总共7条数据，这些数据都会被拉取到同一个task中进行处理；而world和you这两个key分别才对应1条数据，所以另外两个task只要分别处理1条数据即可。此时第一个task的运行时间可能是另外两个task的7倍，而整个stage的运行速度也由运行最慢的那个task所决定。

如何定位导致数据倾斜的代码
数据倾斜只会发生在shuffle过程中。这里给大家罗列一些常用的并且可能会触发shuffle操作的算子：distinct、groupByKey、reduceByKey、aggregateByKey、join、cogroup、repartition等。出现数据倾斜时，可能就是你的代码中使用了这些算子中的某一个所导致的。
more >>

展开全文 >>

Flink和Spark对比

2016-03-16

1、抽象 Abstraction　　Spark中，对于批处理我们有RDD,对于流式，我们有DStream，不过内部实际还是RDD.所以所有的数据表示本质上还是RDD抽象。后面我会重点从不同的角度对比这两者。在Flink中，对于批处理有DataSet，对于流式我们有DataStreams。看起来和Spark类似，他们的不同点在于：　　

（一）DataSet在运行时是表现为运行计划(runtime plans)的　　在Spark中，RDD在运行时是表现为java objects的。通过引入Tungsten，这块有了些许的改变。但是在Flink中是被表现为logical plan(逻辑计划)的，听起来很熟悉？没错，就是类似于Spark中的dataframes。所以在Flink中你使用的类Dataframe api是被作为第一优先级来优化的。但是相对来说在Spark RDD中就没有了这块的优化了。　　Flink中的Dataset，对标Spark中的Dataframe，在运行前会经过优化。在Spark 1.6，dataset API已经被引入Spark了，也许最终会取代RDD 抽象。　　

（二）Dataset和DataStream是独立的API　　在Spark中，所有不同的API，例如DStream，Dataframe都是基于RDD抽象的。但是在Flink中，Dataset和DataStream是同一个公用的引擎之上两个独立的抽象。所以你不能把这两者的行为合并在一起操作，当然，Flink社区目前在朝这个方向努力，但是目前还不能轻易断言最后的结果。

2、内存管理　　一直到1.5版本，Spark都是试用java的内存管理来做数据缓存，明显很容易导致OOM或者gc。所以从1.5开始，Spark开始转向精确的控制内存的使用，这就是tungsten项目了。　　而Flink从第一天开始就坚持自己控制内存试用。这个也是启发了Spark走这条路的原因之一。Flink除了把数据存在自己管理的内存以外，还直接操作二进制数据。在Spark中，从1.5开始，所有的dataframe操作都是直接作用在tungsten的二进制数据上。

3、语言实现　　Spark是用scala来实现的，它提供了Java，Python和R的编程接口。Flink是java实现的，当然同样提供了Scala API所以从语言的角度来看，Spark要更丰富一些。因为我已经转移到scala很久了，所以不太清楚这两者的java api实现情况。

4、API　　Spark和Flink都在模仿scala的collection API.所以从表面看起来，两者都很类似。下面是分别用RDD和DataSet API实现的word count

// Spark wordcountobject WordCount {  def main(args: Array[String]) {    val env = new SparkContext("local","wordCount")    val data = List("hi","how are you","hi")    val dataSet = env.parallelize(data)    val words = dataSet.flatMap(value => value.split("\\s+"))    val mappedWords = (value => (value,1))    val sum = mappedWords.reduceByKey(_+_)    println(sum.collect())  }}


// Flink wordcountobject WordCount {def main(args: Array[String]) {　　val env = ExecutionEnvironment.getExecutionEnvironment　　val data = List("hi","how are you","hi")　　val dataSet = env.fromCollection(data)　　val words = dataSet.flatMap(value => value.split("\\s+"))　　val mappedWords = (value => (value,1))　　val grouped = mappedWords.groupBy(0)　　val sum = grouped.sum(1)　　println(sum.collect())　}}

不知道是偶然还是故意的，API都长得很像，这样很方便开发者从一个引擎切换到另外一个引擎。我感觉以后这种Collection API会成为写data pipeline的标配。

5、Steaming　　Spark把streaming看成是更快的批处理，而Flink把批处理看成streaming的special case。这里面的思路决定了各自的方向，其中两者的差异点有如下这些：实时 vs 近实时的角度：Flink提供了基于每个事件的流式处理机制，所以可以被认为是一个真正的流式计算。它非常像storm的model。而Spark，不是基于事件的粒度，而是用小批量来模拟流式，也就是多个事件的集合。所以Spark被认为是近实时的处理系统。　　

Spark streaming 是更快的批处理，而Flink Batch是有限数据的流式计算。虽然大部分应用对准实时是可以接受的，但是也还是有很多应用需要event level的流式计算。这些应用更愿意选择storm而非Spark streaming，现在，Flink也许是一个更好的选择。流式计算和批处理计算的表示：Spark对于批处理和流式计算，都是用的相同的抽象：RDD，这样很方便这两种计算合并起来表示。而Flink这两者分为了DataSet和DataStream，相比Spark，这个设计算是一个糟糕的设计。对 windowing 的支持：因为Spark的小批量机制，Spark对于windowing的支持非常有限。只能基于process time，且只能对batches来做window。而Flink对window的支持非常到位，且Flink对windowing API的支持是相当给力的，允许基于process time,data time,record 来做windowing。我不太确定Spark是否能引入这些API，不过到目前为止，Flink的windowing支持是要比Spark好的。Steaming这部分Flink胜

6、SQL interface　　目前Spark-sql是Spark里面最活跃的组件之一，Spark提供了类似Hive的sql和Dataframe这种DSL来查询结构化数据，API很成熟，在流式计算中使用很广，预计在流式计算中也会发展得很快。至于Flink，到目前为止，Flink Table API只支持类似DataFrame这种DSL，并且还是处于beta状态，社区有计划增加SQL 的interface，但是目前还不确定什么时候才能在框架中用上。所以这个部分，Spark胜出。

7、外部数据源的整合　　Spark的数据源 API是整个框架中最好的，支持的数据源包括NoSql db,parquet,ORC等，并且支持一些高级的操作，例如predicate push down。Flink目前还依赖map/reduce InputFormat来做数据源聚合。这一场Spark胜

8、Iterative processing　　Spark对机器学习的支持较好，因为可以在Spark中利用内存cache来加速机器学习算法。但是大部分机器学习算法其实是一个有环的数据流，但是在Spark中，实际是用无环图来表示的，一般的分布式处理引擎都是不鼓励试用有环图的。但是Flink这里又有点不一样，Flink支持在runtime中的有环数据流，这样表示机器学习算法更有效而且更有效率。这一点Flink胜出。

9、Stream as platform vs Batch as Platform　　Spark诞生在Map/Reduce的时代，数据都是以文件的形式保存在磁盘中，这样非常方便做容错处理。Flink把纯流式数据计算引入大数据时代，无疑给业界带来了一股清新的空气。这个idea非常类似akka-streams这种。成熟度目前的确有一部分吃螃蟹的用户已经在生产环境中使用Flink了，不过从我的眼光来看，Flink还在发展中，还需要时间来成熟。结论　　目前Spark相比Flink是一个更为成熟的计算框架，但是Flink的很多思路很不错，Spark社区也意识到了这一点，并且逐渐在采用Flink中的好的设计思路，所以学习一下Flink能让你了解一下Streaming这方面的更迷人的思路。

展开全文 >>

Flink on Hadoop 从零搭建

2015-09-25

准备
机器：hostname-0(A机)、hostname-i(Slave机若干)

JAVA: 部署好Java环境，保证环境变量(.bashrc .bash_profile)。

SSH：打通A机到Slave机器的无密码登陆权限(包括A机ssh自己)

Hadoop(2.7*)搭建
1.下包，解压成hadoop目录。
2.在hadoop目录下创建HDFS使用的目录：tmp、hdfs/data、hdfs/name
3.修改hadoop/etc/hadoop目录下配置(使用key/value形式表示XML格式)
core-site.xml //基本配置
fs.defaultFS hdfs://hostname-0:9000 //指定HDFS namenode
hadoop.tmp.dir file:/home/admin/hadoop/tmp //指定临时文件目录
hdfs-site.xml //HDFS配置
dfs.namenode.name.dir file:/home/admin/hadoop/hdfs/name //namenode的存储name信息的文件夹
dfs.datanode.data.dir file:/home/admin/hadoop/hdfs/data //HDFS存储数据的文件夹
dfs.replication 2 //HDFS默认副本数
dfs.namenode.secondary.http-address hostname-0:9001 //HDFS 备用namenode地址
dfs.webhdfs.enabled true //允许HDFS通过web界面访问
mapred-site.xml //MapReduce配置
mapreduce.framework.name yarn //MR通过YARN来调度
mapreduce.jobhistory.address hostname-0:10020 //MR jobHistory server的地址
mapreduce.jobhistory.webapp.address hostname-0:19888 //jobHistory server web地址
yarn-site.xml //YARN配置
yarn.nodemanager.aux-services mapreduce_shuffle //NodeMonitor上运行的附属服务，MR相关
yarn.nodemanager.auxservices.mapreduce.shuffle.class org.apache.hadoop.mapred.ShuffleHandler //同MR相关
yarn.resourcemanager.address hostname-0:8032 //RM对客户端暴露的地址
yarn.resourcemanager.scheduler.address hostname-0:8030 //RM对AppMaster暴露的地址
yarn.resourcemanager.resource-tracker.address hostname-0:8031 //RM对NodeMonitor暴露的地址
yarn.resourcemanager.admin.address hostname-0:8033 //RM对管理员暴露的地址
yarn.resourcemanager.webapp.address hostname-0:8088 //RM对UI暴露的地址
yarn.nodemanager.resource.cpu-vcores 32 //Slave机器的CPU核心数
yarn.nodemanager.resource.memory-mb 102400 //Slave机器的内存总量
yarn.nodemanager.vmem-pmem-ratio 5 //需要设置虚内存比例以防被误杀
slaves //配置slaves
写入slave机的机器名，换行分割
4.Copy hadoop目录到其它slaves机器中。
5.hostname-0上执行bin/hdfs namenode -format，进行HDFS初始化
sbin目录下执行 ./start-all.sh
停止的话，输入命令，sbin/stop-all.sh

more >>

展开全文 >>

协同过滤算法与Mahout推荐系统

2015-08-10

Mahout是什么

Mahout是一个算法库,集成了很多算法。
Apache Mahout 是 Apache Software Foundation（ASF）旗下的一个开源项目，提供一些可扩展的机器学习领域经典算法的实现，旨在帮助开发人员更加方便快捷地创建智能应用程序。
Mahout项目目前已经有了多个公共发行版本。Mahout包含许多实现，包括聚类、分类、推荐过滤、频繁子项挖掘。
通过使用 Apache Hadoop 库，Mahout 可以有效地扩展到Hadoop集群。
Mahout 的创始人 Grant Ingersoll 介绍了机器学习的基本概念，并演示了如何使用 Mahout 来实现文档集群、提出建议和组织内容。

Mahout能做什么

2.1、推荐引擎
在目前采用的机器学习技术中，推荐引擎是最容易被一眼认出来的，也是应用范围最广的。服务商或网站会根据你过去的行为为你推荐书籍、电影或文章。
在部署了推荐系统的电子商务中，亚马逊大概是最有名的。亚马逊基于用户的交易行为和网站记录为你推荐你可能喜欢的商品。
而facebook这样的社交网络则利用推荐技术为你找到最可能尚未关联的朋友。
同时，这一技术也被各大知名国内网站所使用，如腾讯、人人、京东、淘宝。
2.2、聚类
顾名思义，物以类聚，人以群分。聚类是把具有共同属性的物品进行归类。
Google news使用聚类技术通过标题把新闻文章进行分组，从而按照逻辑线索来显示新闻，而并非给出所有新闻的原始列表。
2.3、分类
分类技术决定了一个事物多大程度上从属于某种类别或类型，或者多大程度上具有或者不具有某些属性。与聚类一样，分类无处不在，但更多隐身于幕后。通常这些系统会考察类别中的大量实例，来学习推到出分类的规则。
雅虎邮箱基于用户以前对正常右键和垃圾邮件的报告，以及电子右键自身的特征，来判别到来的消息是否是垃圾邮件。

more >>

展开全文 >>

Kafka整合Spark

2015-07-21

安装zk集群

1.上传zk安装包

2.解压



3.配置（先在一台节点上配置）
	3.1添加一个zoo.cfg配置文件

	$ZOOKEEPER/conf
	mv zoo_sample.cfg zoo.cfg
	
	3.2修改配置文件（zoo.cfg）
		dataDir=/test/zookeeper-3.4.5/data
		
		server.5=test:2888:3888
		server.6=test:2888:3888
		server.7=test:2888:3888
	
	3.3在（dataDir=/test/zookeeper-3.4.5/data）创建一个myid文件，里面内容是server.N中的N（server.2里面内容为2）
		echo "5" > myid
	
	3.4将配置好的zk拷贝到其他节点
		scp -r /test/zookeeper-3.4.5/ test:/test/
		scp -r /test/zookeeper-3.4.5/ test:/test/
	
	3.5注意：在其他节点上一定要修改myid的内容
		在test06应该讲myid的内容改为6 （echo "6" > myid）
		在test07应该讲myid的内容改为7 （echo "7" > myid）
		
4.启动集群
	分别启动zk
		./zkServer.sh start

more >>

展开全文 >>

流式计算Sparkstreaming

2015-06-20

什么是Spark Streaming

Spark Streaming类似于Apache Storm，用于流式数据的处理。根据其官方文档介绍，Spark Streaming有高吞吐量和容错能力强等特点。Spark Streaming支持的数据输入源很多，例如：Kafka、Flume、Twitter、ZeroMQ和简单的TCP套接字等等。数据输入后可以用Spark的高度抽象原语如：map、reduce、join、window等进行运算。而结果也能保存在很多地方，如HDFS，数据库等。另外Spark Streaming也能和MLlib（机器学习）以及Graphx完美融合。

Spark Streaming有哪些优点？

1 易用
2 容错
3 易整合到Spark体系

Spark与Storm的对比

more >>

展开全文 >>

SparkRDD

2015-05-19

弹性分布式数据集RDD

RDD

RDD（Resilient Distributed Dataset）叫做分布式数据集，是Spark中最基本的数据抽象，它代表一个不可变、可分区、里面的元素可并行计算的集合。RDD具有数据流模型的特点：自动容错、位置感知性调度和可伸缩性。RDD允许用户在执行多个查询时显式地将工作集缓存在内存中，后续的查询能够重用工作集，这极大地提升了查询速度。

RDD的属性

1）一组分片（Partition），即数据集的基本组成单位。对于RDD来说，每个分片都会被一个计算任务处理，并决定并行计算的粒度。用户可以在创建RDD时指定RDD的分片个数，如果没有指定，那么就会采用默认值。默认值就是程序所分配到的CPU Core的数目。

2）一个计算每个分区的函数。Spark中RDD的计算是以分片为单位的，每个RDD都会实现compute函数以达到这个目的。compute函数会对迭代器进行复合，不需要保存每次计算的结果。

3）RDD之间的依赖关系。RDD的每次转换都会生成一个新的RDD，所以RDD之间就会形成类似于流水线一样的前后依赖关系。在部分分区数据丢失时，Spark可以通过这个依赖关系重新计算丢失的分区数据，而不是对RDD的所有分区进行重新计算。

4）一个Partitioner，即RDD的分片函数。当前Spark中实现了两种类型的分片函数，一个是基于哈希的HashPartitioner，另外一个是基于范围的RangePartitioner。只有对于于key-value的RDD，才会有Partitioner，非key-value的RDD的Parititioner的值是None。Partitioner函数不但决定了RDD本身的分片数量，也决定了parent RDD Shuffle输出时的分片数量。

5）一个列表，存储存取每个Partition的优先位置（preferred location）。对于一个HDFS文件来说，这个列表保存的就是每个Partition所在的块的位置。按照“移动数据不如移动计算”的理念，Spark在进行任务调度的时候，会尽可能地将计算任务分配到其所要处理数据块的存储位置。

创建RDD

1）由一个已经存在的Scala集合创建。
val rdd1 = sc.parallelize(Array(1,2,3,4,5,6,7,8))

2）由外部存储系统的数据集创建，包括本地的文件系统，还有所有Hadoop支持的数据集，比如HDFS、Cassandra、HBase等
val rdd2 = sc.textFile(“hdfs://node1.test.cn:9000/words.txt”)
more >>

展开全文 >>

实时计算之Kafka

2015-05-13

Kafka是什么

在流式计算中，Kafka一般用来缓存数据，Storm通过消费Kafka的数据进行计算。

消息系统的核心作用就是三点：解耦，异步和并行

KAFKA + STORM +REDIS

Apache Kafka是一个开源消息系统，由Scala写成。是由Apache软件基金会开发的一个开源消息系统项目。
Kafka最初是由LinkedIn开发，并于2011年初开源。2012年10月从Apache Incubator毕业。该项目的目标是为处理实时数据提供一个统一、高通量、低等待的平台。
Kafka是一个分布式消息队列：生产者、消费者的功能。它提供了类似于JMS的特性，但是在设计实现上完全不同，此外它并不是JMS规范的实现。
Kafka对消息保存时根据Topic进行归类，发送消息者称为Producer,消息接受者称为Consumer,此外kafka集群有多个kafka实例组成，每个实例(server)成为broker。
无论是kafka集群，还是producer和consumer都依赖于zookeeper集群保存一些meta信息，来保证系统可用性

JMS是什么

2.1、JMS的基础
JMS是什么：JMS是Java提供的一套技术规范
JMS干什么用：用来异构系统集成通信，缓解系统瓶颈，提高系统的伸缩性增强系统用户体验，使得系统模块化和组件化变得可行并更加灵活
通过什么方式：生产消费者模式（生产者、服务器、消费者）

JMS消息传输模型
点对点模式（一对一，消费者主动拉取数据，消息收到后消息清除）
点对点模型通常是一个基于拉取或者轮询的消息传送模型，这种模型从队列中请求信息，而不是将消息推送到客户端。这个模型的特点是发送到队列的消息被一个且只有一个接收者接收处理，即使有多个消息监听者也是如此。
发布/订阅模式（一对多，数据生产后，推送给所有订阅者）
发布订阅模型则是一个基于推送的消息传送模型。发布订阅模型可以有多种不同的订阅者，临时订阅者只在主动监听主题时才接收消息，而持久订阅者则监听主题的所有消息，即时当前订阅者不可用，处于离线状态。

queue.put（object）数据生产
queue.take(object) 数据消费

more >>

展开全文 >>

SparkSQL

2015-05-11

Spark SQL是Spark用来处理结构化数据的一个模块，它提供了一个编程抽象叫做DataFrame并且作为分布式SQL查询引擎的作用。

Hive是将Hive SQL转换成MapReduce然后提交到集群上执行，大大简化了编写MapReduce的程序的复杂性，由于MapReduce这种计算模型执行效率比较慢。所有Spark SQL的应运而生，它是将Spark SQL转换成RDD，然后提交到集群执行，执行效率非常快！

优点：1.易整合 2.统一的数据访问方式 3.兼容Hive 4.标准的数据连接

什么是DataFrames

与RDD类似，DataFrame也是一个分布式数据容器。然而DataFrame更像传统数据库的二维表格，除了数据以外，还记录数据的结构信息，即schema。同时，与Hive类似，DataFrame也支持嵌套数据类型（struct、array和map）。从API易用性的角度上看，DataFrame API提供的是一套高层的关系操作，比函数式的RDD API要更加友好，门槛更低。由于与R和Pandas的DataFrame类似，Spark DataFrame很好地继承了传统单机数据分析的开发体验。

创建DataFrames
在Spark SQL中SQLContext是创建DataFrames和执行SQL的入口，在spark-1.5.2中已经内置了一个sqlContext

1.在本地创建一个文件，有三列，分别是id、name、age，用空格分隔，然后上传到hdfs上

1	hdfs dfs -put person.txt /

2.在spark shell执行下面命令，读取数据，将每一行的数据使用列分隔符分割

1	val lineRDD = sc.textFile("hdfs://node1.test.cn:9000/person.txt").map(_.split(" "))

more >>

展开全文 >>