大数据
by smallnest

跟着实例学习ZooKeeper的用法： Barrier

分布式Barrier是这样一个类： 它会阻塞所有节点上的等待进程，知道某一个被满足， 然后所有的节点继续进行。

比如赛马比赛中， 等赛马陆续来到起跑线前。 一声令下，所有的赛马都飞奔而出。

栅栏Barrier

DistributedBarrier类实现了栅栏的功能。 它的构造函数如下：

1
2
3
4
public DistributedBarrier(CuratorFramework client, String barrierPath)
Parameters:
client - client
barrierPath - path to use as the barrier

首先你需要设置栅栏，它将阻塞在它上面等待的线程:

1
setBarrier();

然后需要阻塞的线程调用方法等待放行条件:

1
public void waitOnBarrier()

当条件满足时，移除栅栏，所有等待的线程将继续执行：

1
removeBarrier();

异常处理
DistributedBarrier 会监控连接状态，当连接断掉时waitOnBarrier()方法会抛出异常。

看一个例子：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
package com.colobu.zkrecipe.barrier;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.barriers.DistributedBarrier;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.test.TestingServer;
public class DistributedBarrierExample {
	private static final int QTY = 5;
	private static final String PATH = "/examples/barrier";
	public static void main(String[] args) throws Exception {
		try (TestingServer server = new TestingServer()) {
			CuratorFramework client = CuratorFrameworkFactory.newClient(server.getConnectString(), new ExponentialBackoffRetry(1000, 3));
			client.start();
			ExecutorService service = Executors.newFixedThreadPool(QTY);
			DistributedBarrier controlBarrier = new DistributedBarrier(client, PATH);
			controlBarrier.setBarrier();
			
			for (int i = 0; i < QTY; ++i) {
				final DistributedBarrier barrier = new DistributedBarrier(client, PATH);
				final int index = i;
				Callable<Void> task = new Callable<Void>() {
					@Override
					public Void call() throws Exception {
						
						Thread.sleep((long) (3 * Math.random()));
						System.out.println("Client #" + index + " waits on Barrier");
						barrier.waitOnBarrier();
						System.out.println("Client #" + index + " begins");
						return null;
					}
				};
				service.submit(task);
			}
			
			Thread.sleep(10000);
			System.out.println("all Barrier instances should wait the condition");
			
			
			controlBarrier.removeBarrier();
			
			
			service.shutdown();
			service.awaitTermination(10, TimeUnit.MINUTES);
		}
	}
}

这个例子创建了controlBarrier来设置栅栏和移除栅栏。
我们创建了5个线程，在此Barrier上等待。
最后移除栅栏后所有的线程才继续执行。

如果你开始不设置栅栏，所有的线程就不会阻塞住。

双栅栏Double Barrier

双栅栏允许客户端在计算的开始和结束时同步。当足够的进程加入到双栅栏时，进程开始计算， 当计算完成时，离开栅栏。
双栅栏类是DistributedDoubleBarrier。
构造函数为:

1
2
3
4
5
6
7
8
9
10
public DistributedDoubleBarrier(CuratorFramework client,
                                String barrierPath,
                                int memberQty)
Creates the barrier abstraction. memberQty is the number of members in the barrier. When enter() is called, it blocks until
all members have entered. When leave() is called, it blocks until all members have left.
Parameters:
client - the client
barrierPath - path to use
memberQty - the number of members in the barrier

memberQty是成员数量，当enter方法被调用时，成员被阻塞，直到所有的成员都调用了enter。 当leave方法被调用时，它也阻塞调用线程， 知道所有的成员都调用了leave。
就像百米赛跑比赛， 发令枪响， 所有的运动员开始跑，等所有的运动员跑过终点线，比赛才结束。

DistributedBarrier 会监控连接状态，当连接断掉时enter()和leave方法会抛出异常。

例子代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
package com.colobu.zkrecipe.barrier;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.barriers.DistributedBarrier;
import org.apache.curator.framework.recipes.barriers.DistributedDoubleBarrier;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.test.TestingServer;
public class DistributedBarrierExample {
	private static final int QTY = 5;
	private static final String PATH = "/examples/barrier";
	public static void main(String[] args) throws Exception {
		try (TestingServer server = new TestingServer()) {
			CuratorFramework client = CuratorFrameworkFactory.newClient(server.getConnectString(), new ExponentialBackoffRetry(1000, 3));
			client.start();
			ExecutorService service = Executors.newFixedThreadPool(QTY);
			for (int i = 0; i < QTY; ++i) {
				final DistributedDoubleBarrier barrier = new DistributedDoubleBarrier(client, PATH, QTY);
				final int index = i;
				Callable<Void> task = new Callable<Void>() {
					@Override
					public Void call() throws Exception {
						
						Thread.sleep((long) (3 * Math.random()));
						System.out.println("Client #" + index + " enters");
						barrier.enter();
						System.out.println("Client #" + index + " begins");
						Thread.sleep((long) (3000 * Math.random()));
						barrier.leave();
						System.out.println("Client #" + index + " left");
						return null;
					}
				};
				service.submit(task);
			}
			
			
			service.shutdown();
			service.awaitTermination(10, TimeUnit.MINUTES);
		}
	}
}

分布式Barrier是这样一个类： 它会阻塞所有节点上的等待进程，知道某一个被满足， 然后所有的节点继续进行。

阅读全文

大数据
by smallnest

跟着实例学习ZooKeeper的用法： Leader选举

ZooKeeper官方给出了使用zookeeper的几种用途。

  • Leader Election
  • Barriers
  • Queues
  • Locks
  • Two-phased Commit
  • 其它应用如Name Service, Configuration, Group Membership

在实际使用ZooKeeper开发中，我们最常用的是Apache Curator。 它由Netflix公司贡献给Apache，目前版本2.7.0。
相信你在使用ZK API开发时会遇到让人头疼的几个问题，ZK连接管理、SESSION失效等一些异常问题的处理，Curator替我们解决了这些问题，通过对ZK连接状态的监控来做出相应的重连等操作，并触发事件。
更好的地方是Curator对ZK的一些应用场景提供了非常好的实现，而且有很多扩充，这些都符合ZK使用规范。
它的主要组件为：

  • Recipes， ZooKeeper的系列recipe实现, 基于 Curator Framework.
  • Framework， 封装了大量ZooKeeper常用API操作，降低了使用难度, 基于Zookeeper增加了一些新特性，对ZooKeeper链接的管理，对链接丢失自动重新链接。
  • Utilities，一些ZooKeeper操作的工具类包括ZK的集群测试工具路径生成等非常有用，在Curator-Client包下org.apache.curator.utils。
  • Client，ZooKeeper的客户端API封装，替代官方 ZooKeeper class，解决了一些繁琐低级的处理，提供一些工具类。
  • Errors，异常处理, 连接异常等
  • Extensions，对curator-recipes的扩展实现，拆分为curator-:stuck_out_tongue_closed_eyes:iscovery和curator-:stuck_out_tongue_closed_eyes:iscovery-server提供基于RESTful的Recipes WEB服务.

Recipe 词典的意思是食谱,配方,美食菜谱,烹饪法， 延伸用法：某项计划或步骤来取得预先给定的结果。 在计算机领域没有合适的汉语对应，如果把ZooKeeper看成菜的话，recipe就相当于菜谱， 比如麻婆豆腐， 宫保鸡丁。

由于内容较多， 将会分成多篇文章进行介绍。

阅读全文

大数据
by smallnest

Spark SQL 初探： 使用大数据分析2000万数据

去年网上曾放出个2000W的开房记录的数据库， 不知真假。 最近在学习Spark， 所以特意从网上找来数据测试一下， 这是一个绝佳的大数据素材。 如果数据涉及到个人隐私，请尽快删除， 本站不提供此类数据。你可以写个随机程序生成2000W的测试数据， 以CSV格式。

Spark是UC Berkeley AMP lab所开源的类Hadoop MapReduce的通用的并行计算框架，Spark基于map reduce算法实现的分布式计算，拥有Hadoop MapReduce所具有的优点；但不同于MapReduce的是Job中间输出和结果可以保存在内存中，从而不再需要读写HDFS，因此Spark能更好地适用于数据挖掘与机器学习等需要迭代的map reduce的算法。

Spark是一个高效的分布式计算系统，相比Hadoop，它在性能上比Hadoop要高100倍。Spark提供比Hadoop更上层的API，同样的算法在Spark中实现往往只有Hadoop的1/10或者1/100的长度。Shark类似“SQL on Spark”，是一个在Spark上数据仓库的实现，在兼容Hive的情况下，性能最高可以达到Hive的一百倍。

Apache Spark 是在 Scala 语言中实现的，它将 Scala 用作其应用程序框架。与 Hadoop 不同，Spark 和 Scala 能够紧密集成，其中的 Scala 可以像操作本地集合对象一样轻松地操作分布式数据集。

2014年处， Apache 基金会宣布旗下的 Apache Spark 项目成为基金会的顶级项目，拥有顶级域名 http://spark.apache.org/。 Spark 的用户包括：阿里巴巴、Cloudera、Databricks、IBM、英特尔和雅虎等知名厂商。

Spark SQL是支持在Spark中使用Sql、HiveSql、Scaca中的关系型查询表达式。它的核心组件是一个新增的RDD类型SchemaRDD，它把行对象用一个Schema来描述行里面的所有列的数据类型，它就像是关系型数据库里面的一张表。它可以从原有的RDD创建，也可以是Parquet文件，最重要的是它可以支持用HiveQL从hive里面读取数据。

在2014年7月1日的Spark Summit上，Databricks宣布终止对Shark的开发，将重点放到Spark SQL上。在会议上，Databricks表示，Shark更多是对Hive的改造，替换了Hive的物理执行引擎，因此会有一个很快的速度。然而，不容忽视的是，Shark继承了大量的Hive代码，因此给优化和维护带来了大量的麻烦。随着性能优化和先进分析整合的进一步加深，基于MapReduce设计的部分无疑成为了整个项目的瓶颈。 详细内容请参看 Shark, Spark SQL, Hive on Spark, and the future of SQL on Spark

当前Spark SQL还处于alpha阶段，一些API在将将来的版本中可能会有所改变。

我也翻译几篇重要的Spark文档，你可以在我的网站找到。 Spark翻译文档

本文主要介绍了下面几个知识点：

  • Spark读取文件夹的文件
  • Spark filter和map使用
  • Spark sql语句调用
  • 自定义Spark sql的函数

    阅读全文

大数据
by smallnest

Spark 配置指南

Spark可以在三个地方配置系统:

  • Spark属性控制大部分的应用参数。 这些属性可以通过SparkConf对象, 或者Java系统属性.
  • 环境变量可以为每台机器配置，比如IP地址， 通过每个节点上的conf/spark-env.sh脚本.
  • 可同通过log4j.properties配置日志.

阅读全文