首页 文章 精选 留言 我的

精选列表

搜索[guide]，共2033篇文章
优秀的个人博客,低调大师

Application Architecture Guide 2.0

最近比较有时间了，想翻译点有意义的东西和大家分享下同时也提高下自己的英语水平。 有什么不对的地方就请大家拍砖赐教了！ 应用程序架构指导2 引言 -J.D. Meier, Alex Homer, David Hill,Jason Taylor,Prashant Bansode, Lonnie Wall, Rob Boucher Jr, Akshay Bogawat 前言 应用程序架构指导的目的是为了改善你在微软平台上开发的效率。主要的读者包括解决方案架构师和团队开发领导。本文为在.Net平台上建立的应用程序提供架构和设计上的指导。本文主要集中在一般类型的应用程序上，把应用程序的功能分割为几层，组件和服务，并介绍它们的共通的设计特点。 这个指导是给予任务的，并且会按照架构和设计的主要特点分别介绍。本文既可以作为参考，也可以的从开始学习到最后。这个指导被分为一下四部分： l第一部分：“基础”提供架构和设计方面的基础知识，一边理解架构设计的技巧和策略。 l第二部分：“设计”提供设计主要的可以用于任何类型的应用程序或者程序中的某一层的原则和实践，包括如何设计联系和服务。 l第三部分：“分层”体统架构和设计方式，以及每一层的实践，包括展示、业务、服务和数据访问。 l第四部分：为每一种应用程序原型提供模式和设计框架，包括服务应用、Web应用、富客户端应用、RIA应用 我们为什么写这个指南 我们写作这个指南是为了完成一下目标： •帮助你在.NET平台下设计更有效率的架构 •帮助你选择正确的技术 •帮助你采取更有效的工程决策 •帮助你选择适合的策略和模式 •帮助你选择相关度模式和实践解决方案。 范围 这个原则提供了在.NET平台下应用程序架构方面的原则、模式和实践。 这是个原则为基础的方式。这个指南的范围请参看（图1） （图1） 欢迎加群互相学习，共同进步。QQ群：iOS: 58099570 | Android: 572064792 | Nodejs：329118122 做人要厚道，转载请注明出处！ 本文转自张昺华-sky博客园博客，原文链接：http://www.cnblogs.com/sunshine-anycall/archive/2008/12/22/1359696.html ，如需转载请自行联系原作者
2017-11-07
优秀的个人博客,低调大师

Flink DataStream API Programming Guide

Example Program The following program is a complete, working example of streaming window word count application, that counts the words coming from a web socket in 5 second windows. public class WindowWordCount { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); DataStream<Tuple2<String, Integer>> dataStream = env .socketTextStream("localhost", 9999) .flatMap(new Splitter()) .keyBy(0) .timeWindow(Time.of(5, TimeUnit.SECONDS)) .sum(1); dataStream.print(); env.execute("Window WordCount"); } public static class Splitter implements FlatMapFunction<String, Tuple2<String, Integer>> { @Override public void flatMap(String sentence, Collector<Tuple2<String, Integer>> out) throws Exception { for (String word: sentence.split(" ")) { out.collect(new Tuple2<String, Integer>(word, 1)); } } } } Flink应用的代码结构如下， Flink DataStream programs look like regular Java programs with amain()method. Each program consists of the same basic parts: Obtaining aStreamExecutionEnvironment, Connecting to data stream sources, Specifying transformations on the data streams, Specifying output for the processed data, Executing the program. 以这个例子，说明 首先会创建socketTextStream，从socket读入text流 接着是个flatMap，和map的不同是，map，1->1，而flatMap为1->n，而这个splitter就是将text用“”分割，将每个word作为一个tuple输出 最后，keyBy产生一个有key的tuple流，这里是以word为key 基于5s的timeWindow，对后面的计数进行sum 最终，output是print Transformations 太常用的就不列了 ============================================================================================== Reduce KeyedStream → DataStream A "rolling" reduce on a keyed data stream. Combines the current element with the last reduced value and emits the new value. keyedStream.reduce(new ReduceFunction<Integer>() { @Override public Integer reduce(Integer value1, Integer value2) throws Exception { return value1 + value2; } }); Fold KeyedStream → DataStream A "rolling" fold on a keyed data stream with an initial value. Combines the current element with the last folded value and emits the new value. A fold function that, when applied on the sequence (1,2,3,4,5), emits the sequence "start-1", "start-1-2", "start-1-2-3", ... DataStream<String> result = keyedStream.fold("start", new FoldFunction<Integer, String>() { @Override public String fold(String current, Integer value) { return current + "-" + value; } }); Fold和reduce的区别，fold可以有个初始值，而且foldfunciton可以将一种类型fold到另一种类型 而reduce function，只能是一种类型 Aggregations KeyedStream → DataStream Rolling aggregations on a keyed data stream. The difference between min and minBy is that min returns the minimun value, whereas minBy returns the element that has the minimum value in this field (same for max and maxBy). keyedStream.sum(0); keyedStream.sum("key"); keyedStream.min(0); keyedStream.min("key"); keyedStream.max(0); keyedStream.max("key"); keyedStream.minBy(0); keyedStream.minBy("key"); keyedStream.maxBy(0); keyedStream.maxBy("key"); 可以认为是特殊的reduce 不带by，只是返回value 带by，返回整个element ============================================================================================= Union DataStream* → DataStream Union of two or more data streams creating a new stream containing all the elements from all the streams. Node: If you union a data stream with itself you will get each element twice in the resulting stream. dataStream.union(otherStream1, otherStream2, ...); Connect DataStream,DataStream → ConnectedStreams "Connects" two data streams retaining their types. Connect allowing for shared state between the two streams. DataStream<Integer> someStream = //... DataStream<String> otherStream = //... ConnectedStreams<Integer, String> connectedStreams = someStream.connect(otherStream); connect就是两个不同type的流可以共享一个流，tuple可以同时拿到来自两个流的数据 CoMap, CoFlatMap ConnectedStreams → DataStream Similar to map and flatMap on a connected data stream connectedStreams.map(new CoMapFunction<Integer, String, Boolean>() { @Override public Boolean map1(Integer value) { return true; } @Override public Boolean map2(String value) { return false; } }); Split DataStream → SplitStream Split the stream into two or more streams according to some criterion. SplitStream<Integer> split = someDataStream.split(new OutputSelector<Integer>() { @Override public Iterable<String> select(Integer value) { List<String> output = new ArrayList<String>(); if (value % 2 == 0) { output.add("even"); } else { output.add("odd"); } return output; } }); Select SplitStream → DataStream Select one or more streams from a split stream. SplitStream<Integer> split; DataStream<Integer> even = split.select("even"); DataStream<Integer> odd = split.select("odd"); DataStream<Integer> all = split.select("even","odd"); ==================================================================================== Project DataStream → DataStream Selects a subset of fields from the tuples DataStream<Tuple3<Integer, Double, String>> in = // [...] DataStream<Tuple2<String, Integer>> out = in.project(2,0); =========================================================================================== Window KeyedStream → WindowedStream Windows can be defined on already partitioned KeyedStreams. Windows group the data in each key according to some characteristic (e.g., the data that arrived within the last 5 seconds). Seewindowsfor a complete description of windows. dataStream.keyBy(0).window(TumblingTimeWindows.of(Time.of(5, TimeUnit.SECONDS))); // Last 5 seconds of data 基于keyedStream的window WindowAll DataStream → AllWindowedStream Windows can be defined on regular DataStreams. Windows group all the stream events according to some characteristic (e.g., the data that arrived within the last 5 seconds). Seewindowsfor a complete description of windows. WARNING:This is in many cases anon-paralleltransformation. All records will be gathered in one task for the windowAll operator. dataStream.windowAll(TumblingTimeWindows.of(Time.of(5, TimeUnit.SECONDS))); // Last 5 seconds of data 主要，由于没有key，所以如果要对all做transform，是无法parallel的，只能在一个task里面做 Window Apply WindowedStream → DataStream AllWindowedStream → DataStream Applies a general function to the window as a whole. Below is a function that manually sums the elements of a window. Note:If you are using a windowAll transformation, you need to use an AllWindowFunction instead. Window Reduce WindowedStream → DataStream Applies a functional reduce function to the window and returns the reduced value. Aggregations on windows WindowedStream → DataStream Aggregates the contents of a window. The difference between min and minBy is that min returns the minimun value, whereas minBy returns the element that has the minimum value in this field (same for max and maxBy). windowedStream.sum(0); windowedStream.sum("key"); Window Join DataStream,DataStream → DataStream Join two data streams on a given key and a common window. dataStream.join(otherStream) .where(0).equalTo(1) .window(TumblingTimeWindows.of(Time.of(3, TimeUnit.SECONDS))) .apply (new JoinFunction () {...}); Physical partitioning 类似storm的group方式，可以自己配置 Hash partitioning， 等同于 groupby field DataStream → DataStream Identical to keyBy but returns a DataStream instead of a KeyedStream. dataStream.partitionByHash("someKey"); dataStream.partitionByHash(0); Custom partitioning DataStream → DataStream Uses a user-defined Partitioner to select the target task for each element. dataStream.partitionCustom(new Partitioner(){...}, "someKey"); dataStream.partitionCustom(new Partitioner(){...}, 0); Random partitioning，等同于shuffle DataStream → DataStream Partitions elements randomly according to a uniform distribution. dataStream.partitionRandom(); Rebalancing (Round-robin partitioning) DataStream → DataStream Partitions elements round-robin, creating equal load per partition. Useful for performance optimization in the presence of data skew. dataStream.rebalance(); 这个保证数据不会skew，round-robin就是每个一条，轮流来 Broadcasting，等同于globle DataStream → DataStream Broadcasts elements to every partition. dataStream.broadcast(); Task chaining and resource groups Chaining two subsequent transformations means co-locating them within the same thread for better performance. Flink by default chains operators if this is possible (e.g., two subsequent map transformations). The API gives fine-grained control over chaining if desired: A resource group is a slot in Flink, seeslots. You can manually isolate operators in separate slots if desired. Start new chain Begin a new chain, starting with this operator. The two mappers will be chained, and filter will not be chained to the first mapper. someStream.filter(...).map(...).startNewChain().map(...); 注意startNewChain是应用于，左边的那个operator，所以上面从第一个map开始start new chain Disable chaining Do not chain the map operator someStream.map(...).disableChaining(); Start a new resource group Start a new resource group containing the map and the subsequent operators. someStream.map(...).startNewResourceGroup(); 意思就是他们share同一个slot？ Isolate resources Isolate the operator in its own slot. someStream.map(...).isolateResources(); 使用独立的slot Execution Configuration 只有下面两个和batch的配置不同， Parameters in theExecutionConfigthat pertain specifically to the DataStream API are: enableTimestamps()/disableTimestamps(): Attach a timestamp to each event emitted from a source.areTimestampsEnabled()returns the current value. setAutoWatermarkInterval(long milliseconds): Set the interval for automatic watermark emission. You can get the current value withlong getAutoWatermarkInterval() Debugging A LocalEnvironment is created and used as follows: final StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment(); DataStream<String> lines = env.addSource(/* some source */); // build your program env.execute(); Collection data sources can be used as follows: final StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment(); // Create a DataStream from a list of elements DataStream<Integer> myInts = env.fromElements(1, 2, 3, 4, 5); // Create a DataStream from any Java collection List<Tuple2<String, Integer>> data = ... DataStream<Tuple2<String, Integer>> myTuples = env.fromCollection(data); // Create a DataStream from an Iterator Iterator<Long> longIt = ... DataStream<Long> myLongs = env.fromCollection(longIt, Long.class); Flink also provides a sink to collect DataStream results for testing and debugging purposes. It can be used as follows: import org.apache.flink.contrib.streaming.DataStreamUtils DataStream<Tuple2<String, Integer>> myResult = ... Iterator<Tuple2<String, Integer>> myOutput = DataStreamUtils.collect(myResult) Windows Working with Time 3种时间， Processing time，真正的处理时间 Event time， 事件真正发生的时间 Ingestion time，数据进入flink时间，在data source env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime); env.setStreamTimeCharacteristic(TimeCharacteristic.IngestionTime); env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime); 默认是用processing 时间， 如果要用event time，you need to follow four steps: Setenv.setStreamTimeCharacteristic(TimeCharacteristic.EventTime) UseDataStream.assignTimestamps(...)in order to tell Flink how timestamps relate to events (e.g., which record field is the timestamp) SetenableTimestamps(), as well the interval for watermark emission (setAutoWatermarkInterval(long milliseconds)) inExecutionConfig. For example, assume that we have a data stream of tuples, in which the first field is the timestamp (assigned by the system that generates these data streams), and we know that the lag between the current processing time and the timestamp of an event is never more than 1 second: DataStream<Tuple4<Long,Integer,Double,String>> stream = //... stream.assignTimestamps(new TimestampExtractor<Tuple4<Long,Integer,Double,String>>{ @Override public long extractTimestamp(Tuple4<Long,Integer,Double,String> element, long currentTimestamp) { return element.f0; } @Override public long extractWatermark(Tuple4<Long,Integer,Double,String> element, long currentTimestamp) { return element.f0 - 1000; } @Override public long getCurrentWatermark() { return Long.MIN_VALUE; } }); Basic Window Constructs Tumbling time window，非滑动 KeyedStream → WindowedStream Defines a window of 5 seconds, that "tumbles". keyedStream.timeWindow(Time.of(5, TimeUnit.SECONDS)); Sliding time window，滑动 KeyedStream → WindowedStream Defines a window of 5 seconds, that "slides" by 1 seconds. keyedStream.timeWindow(Time.of(5, TimeUnit.SECONDS), Time.of(1, TimeUnit.SECONDS)); Tumbling count window KeyedStream → WindowedStream keyedStream.countWindow(1000); Sliding count window KeyedStream → WindowedStream keyedStream.countWindow(1000, 100) Advanced Window Constructs The general recipe for building a custom window is to specify (1) aWindowAssigner, (2) aTrigger(optionally), and (3) anEvictor(optionally). 上面的如timeWindow，是封装好的，而如果用advanced构建方式，需要3步， 1. 首先是WindowAssigner，主要是滑动和非滑动两类，解决主要的是where的问题 Global window KeyedStream → WindowedStream All incoming elements of a given key are assigned to the same window. The window does not contain a default trigger, hence it will never be triggered if a trigger is not explicitly specified. stream.window(GlobalWindows.create()); 用于count window Tumbling time windows KeyedStream → WindowedStream stream.window(TumblingTimeWindows.of(Time.of(1, TimeUnit.SECONDS))); The window comes with a default trigger. For event/ingestion time, a window is triggered when a watermark with value higher than its end-value is received, whereas for processing time when the current processing time exceeds its current end value. 默认的trigger， 先理解watermark的含义：当我收到一个watermark时，表示我不可能收到event time 小于该water mark的数据 所以我收到的water mark都大于我window的结束时间，说明，window的数据已经到齐了，可以触发trigger Sliding time windows KeyedStream → WindowedStream stream.window(SlidingTimeWindows.of(Time.of(5, TimeUnit.SECONDS), Time.of(1, TimeUnit.SECONDS))); 默认的trigger与上同， 2. 第二步，是定义trigger，何时触发，解决的是when的问题 TheTriggerspecifies when the function that comes after the window clause (e.g.,sum,count) is evaluated (“fires”) for each window. If a trigger is not specified, a default trigger for each window type is used (that is part of the definition of theWindowAssigner). Processing time trigger A window is fired when the current processing time exceeds its end-value. The elements on the triggered window are henceforth discarded. windowedStream.trigger(ProcessingTimeTrigger.create()); Watermark trigger A window is fired when a watermark with value that exceeds the window's end-value has been received. The elements on the triggered window are henceforth discarded. windowedStream.trigger(EventTimeTrigger.create()); Continuous processing time trigger A window is periodically considered for being fired (every 5 seconds in the example). The window is actually fired only when the current processing time exceeds its end-value. The elements on the triggered window are retained. windowedStream.trigger(ContinuousProcessingTimeTrigger.of(Time.of(5, TimeUnit.SECONDS))); Continuous watermark time trigger A window is periodically considered for being fired (every 5 seconds in the example). A window is actually fired when a watermark with value that exceeds the window's end-value has been received. The elements on the triggered window are retained. windowedStream.trigger(ContinuousEventTimeTrigger.of(Time.of(5, TimeUnit.SECONDS))); 这个和上面的不同，在于，window在触发后，不会被discard，而是会保留，并且每隔一段时间会反复的触发 Count trigger A window is fired when it has more than a certain number of elements (1000 below). The elements of the triggered window are retained. windowedStream.trigger(CountTrigger.of(1000)); 按count触发，window会被保留 Purging trigger Takes any trigger as an argument and forces the triggered window elements to be "purged" (discarded) after triggering. windowedStream.trigger(PurgingTrigger.of(CountTrigger.of(1000))); 上面有些trigger是会retain数据的，如果你想discard，怎么搞？ 用PurgingTrigger Delta trigger A window is periodically considered for being fired (every 5000 milliseconds in the example). A window is actually fired when the value of the last added element exceeds the value of the first element inserted in the window according to a `DeltaFunction`. windowedStream.trigger(new DeltaTrigger.of(5000.0, new DeltaFunction<Double>() { @Override public double getDelta (Double old, Double new) { return (new - old > 0.01); } })); Delta trigger，即，每次会通过getDelta比较新来的值和旧值的delta，当delta大于定义的阈值时，就会fire 3. 最后，指定Evictor After the trigger fires, and before the function (e.g.,sum,count) is applied to the window contents, an optionalEvictorremoves some elements from the beginning of the window before the remaining elements are passed on to the function. 说白了，当windows被触发时，我们可以选取部分数据进行处理， evictor，清除者，即清除部分数据，保留你想要的 Time evictor Evict all elements from the beginning of the window, so that elements from end-value - 1 second until end-value are retained (the resulting window size is 1 second). triggeredStream.evictor(TimeEvictor.of(Time.of(1, TimeUnit.SECONDS))); Count evictor Retain 1000 elements from the end of the window backwards, evicting all others. triggeredStream.evictor(CountEvictor.of(1000)); 逻辑是保留，而不是清除，比如CountEvictor.of(1000)是保留最后1000个，有点不好理解 Delta evictor Starting from the beginning of the window, evict elements until an element with value lower than the value of the last element is found (by a threshold and a DeltaFunction). triggeredStream.evictor(DeltaEvictor.of(5000, new DeltaFunction<Double>() { public double (Double oldValue, Double newValue) { return newValue - oldValue; } })); Recipes for Building Windows 下面给出一些window定义的例子，理解一下，例子给的太简单 Windows on Unkeyed Data Streams window，也可以用于unkeyed的数据流， 不同，是在window后面加上all， Tumbling time window all DataStream → WindowedStream Defines a window of 5 seconds, that "tumbles". This means that elements are grouped according to their timestamp in groups of 5 second duration, and every element belongs to exactly one window. The notion of time used is controlled by the StreamExecutionEnvironment. nonKeyedStream.timeWindowAll(Time.of(5, TimeUnit.SECONDS)); Sliding time window all DataStream → WindowedStream Defines a window of 5 seconds, that "slides" by 1 seconds. This means that elements are grouped according to their timestamp in groups of 5 second duration, and elements can belong to more than one window (since windows overlap by at least 4 seconds) The notion of time used is controlled by the StreamExecutionEnvironment. nonKeyedStream.timeWindowAll(Time.of(5, TimeUnit.SECONDS), Time.of(1, TimeUnit.SECONDS)); Tumbling count window all DataStream → WindowedStream Defines a window of 1000 elements, that "tumbles". This means that elements are grouped according to their arrival time (equivalent to processing time) in groups of 1000 elements, and every element belongs to exactly one window. nonKeyedStream.countWindowAll(1000) Sliding count window all DataStream → WindowedStream Defines a window of 1000 elements, that "slides" every 100 elements. This means that elements are grouped according to their arrival time (equivalent to processing time) in groups of 1000 elements, and every element can belong to more than one window (as windows overlap by at least 900 elements). nonKeyedStream.countWindowAll(1000, 100) Working with State All transformations in Flink may look like functions (in the functional processing terminology), but are in fact stateful operators. You can makeeverytransformation (map,filter, etc) stateful by declaring local variables or using Flink’s state interface. You can register any local variable asmanagedstate by implementing an interface. In this case, and also in the case of using Flink’s native state interface, Flink will automatically take consistent snapshots of your state periodically, and restore its value in the case of a failure. The end effect is that updates to any form of state are the same under failure-free execution and execution under failures. First, we look at how to make local variables consistent under failures, and then we look at Flink’s state interface. By default state checkpoints will be stored in-memory at the JobManager. For proper persistence of large state, Flink supports storing the checkpoints on file systems (HDFS, S3, or any mounted POSIX file system), which can be configured in theflink-conf.yamlor viaStreamExecutionEnvironment.setStateBackend(…). 这块是Flink流式处理的核心价值，可以方便的checkpoint的local state，有几种方式，后面会具体说； 默认情况下，这些checkpoints 是存储在JobManager的内存中的，当然也可以配置checkpoint到文件系统 Checkpointing Local Variables 这个比较好理解 Local variables can be checkpointed by using theCheckpointedinterface. When the user-defined function implements theCheckpointedinterface, thesnapshotState(…)andrestoreState(…)methods will be executed to draw and restore function state. public class CounterSum implements ReduceFunction<Long>, Checkpointed<Long> { // persistent counter private long counter = 0; @Override public Long reduce(Long value1, Long value2) { counter++; return value1 + value2; } // regularly persists state during normal operation @Override public Serializable snapshotState(long checkpointId, long checkpointTimestamp) { return counter; } // restores state on recovery from failure @Override public void restoreState(Long state) { counter = state; } } 如上，只是实现snapshotState和restoreState，就可以对local变量counter实现checkpoint，这个很好理解 n addition to that, user functions can also implement theCheckpointNotifierinterface to receive notifications on completed checkpoints via thenotifyCheckpointComplete(long checkpointId)method. Note that there is no guarantee for the user function to receive a notification if a failure happens between checkpoint completion and notification. The notifications should hence be treated in a way that notifications from later checkpoints can subsume missing notifications.、 除此，还能实现CheckpointNotifier，这样当完成checkpoints时，会调用notifyCheckpointComplete，但不能保证一定触发 Using the Key/Value State Interface 这个是显式调用state interface The state interface gives access to key/value states, which are a collection of key/value pairs. Because the state is partitioned by the keys (distributed accross workers), it can only be used on theKeyedStream, created viastream.keyBy(…)(which means also that it is usable in all types of functions on keyed windows). The handle to the state can be obtained from the function’sRuntimeContext. The state handle will then give access to the value mapped under the key of the current record or window - each key consequently has its own value. The following code sample shows how to use the key/value state inside a reduce function. When creating the state handle, one needs to supply a name for that state (a function can have multiple states of different types), the type of the state (used to create efficient serializers), and the default value (returned as a value for keys that do not yet have a value associated). public class CounterSum implements RichReduceFunction<Long> { /** The state handle */ private OperatorState<Long> counter; @Override public Long reduce(Long value1, Long value2) { counter.update(counter.value() + 1); return value1 + value2; } @Override public void open(Configuration config) { counter = getRuntimeContext().getKeyValueState("myCounter", Long.class, 0L); } } State updated by this is usually kept locally inside the flink process (unless one configures explicitly an external state backend). This means that lookups and updates areprocess localand this very fast. The important implication of having the keys set implicitly is that it forces programs to group the stream by key (via thekeyBy()function), making the key partitioning transparent to Flink. That allows the system to efficiently restore and redistribute keys and state. The Scala API has shortcuts that for statefulmap()orflatMap()functions onKeyedStream, which give the state of the current key as an option directly into the function, and return the result with a state update: val stream: DataStream[(String, Int)] = ... val counts: DataStream[(String, Int)] = stream .keyBy(_._1) .mapWithState((in: (String, Int), count: Option[Int]) => count match { case Some(c) => ( (in._1, c), Some(c + in._2) ) case None => ( (in._1, 0), Some(in._2) ) }) State Checkpoints in Iterative Jobs Flink currently only provides processing guarantees for jobs without iterations. Enabling checkpointing on an iterative job causes an exception. In order to force checkpointing on an iterative program the user needs to set a special flag when enabling checkpointing:env.enableCheckpointing(interval, force = true). Please note that records in flight in the loop edges (and the state changes associated with them) will be lost during failure. 对于iterative，即有环的case，做checkpoint更加复杂点，并且恢复后，会丢失中间过程，比如n次迭代，执行到n-1次，失败，还是要从1开始 Iterations For example, here is program that continuously subtracts 1 from a series of integers until they reach zero: DataStream<Long> someIntegers = env.generateSequence(0, 1000); IterativeStream<Long> iteration = someIntegers.iterate(); DataStream<Long> minusOne = iteration.map(new MapFunction<Long, Long>() { @Override public Long map(Long value) throws Exception { return value - 1 ; } }); DataStream<Long> stillGreaterThanZero = minusOne.filter(new FilterFunction<Long>() { @Override public boolean filter(Long value) throws Exception { return (value > 0); } }); iteration.closeWith(stillGreaterThanZero); DataStream<Long> lessThanZero = minusOne.filter(new FilterFunction<Long>() { @Override public boolean filter(Long value) throws Exception { return (value <= 0); } }); 这个直接看例子， 首先，someIntegers是一个由0到1000的DataStream 对于每个tuple，都需要迭代的执行一个map function，在这儿，会不断减一 什么时候结束， 根据iteration.closeWith，closeWith后面是一个filter，如果filter返回为true，这个tuple就继续iterate，如果返回为false，就close iterate 而最后的lessThanZero是someIntegers经过iterate后，最终产生的输出DataStream Connectors Connectors provide code for interfacing with various third-party systems. Currently these systems are supported: Apache Kafka(sink/source) Elasticsearch(sink) Hadoop FileSystem(sink) RabbitMQ(sink/source) Twitter Streaming API(source) To run an application using one of these connectors, additional third party components are usually required to be installed and launched, e.g. the servers for the message queues. Further instructions for these can be found in the corresponding subsections.Docker containersare also provided encapsulating these services to aid users getting started with connectors. 只看下kafka， Then, import the connector in your maven project: <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-connector-kafka</artifactId> <version>0.10.2</version> </dependency> 使用的例子， Properties properties = new Properties(); properties.setProperty("bootstrap.servers", "localhost:9092"); properties.setProperty("zookeeper.connect", "localhost:2181"); properties.setProperty("group.id", "test"); DataStream<String> stream = env .addSource(new FlinkKafkaConsumer082<>("topic", new SimpleStringSchema(), properties)) .print(); 如何fault tolerance？ With Flink’s checkpointing enabled, the Flink Kafka Consumer will consume records from a topic and periodically checkpoint all its Kafka offsets, together with the state of other operations, in a consistent manner. In case of a job failure, Flink will restore the streaming program to the state of the latest checkpoint and re-consume the records from Kafka, starting from the offsets that where stored in the checkpoint. 原理就是会和其他state一起把所有的kafka partition的offset都checkpoint下来，这样恢复的时候，可以从这些offset开始读； To use fault tolerant Kafka Consumers, checkpointing of the topology needs to be enabled at the execution environment: final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.enableCheckpointing(5000); // checkpoint every 5000 msecs If checkpointing is not enabled, the Kafka consumer will periodically commit the offsets to Zookeeper. 由于用的是simple consumer，所以就算不开checkpoint，offset也要被记录；这里使用通常的做法把kafka的offset记录到zookeeper 也可以把数据写入kafka，FlinkKafkaProducer TheFlinkKafkaProducerwrites data to a Kafka topic. The producer can specify a custom partitioner that assigns recors to partitions. tream.addSink(new FlinkKafkaProducer<String>("localhost:9092", "my-topic", new SimpleStringSchema()));
2017-04-06
优秀的个人博客,低调大师

Flink DataSet API Programming Guide

Example Program 编程的风格和spark很类似， ExecutionEnvironment -- SparkContext DataSet – RDD Transformations 这里用Java的接口，所以传入function需要用FlatMapFunction类对象 public class WordCountExample { public static void main(String[] args) throws Exception { final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment(); DataSet<String> text = env.fromElements( "Who's there?", "I think I hear them. Stand, ho! Who's there?"); DataSet<Tuple2<String, Integer>> wordCounts = text .flatMap(new LineSplitter()) .groupBy(0) .sum(1); wordCounts.print(); } public static class LineSplitter implements FlatMapFunction<String, Tuple2<String, Integer>> { @Override public void flatMap(String line, Collector<Tuple2<String, Integer>> out) { for (String word : line.split(" ")) { out.collect(new Tuple2<String, Integer>(word, 1)); } } } } Specifying Keys 如何定义key， 1. 用tuple的index，如下用tuple的第一个和第二个做联合key DataSet<Tuple3<Integer,String,Long>> input = // [...] DataSet<Tuple3<Integer,String,Long> grouped = input .groupBy(0,1) .reduce(/*do something*/); 2. 对于POJO对象，使用Field Expressions // some ordinary POJO (Plain old Java Object) public class WC { public String word; public int count; } DataSet<WC> words = // [...] DataSet<WC> wordCounts = words.groupBy("word").reduce(/*do something*/); 3. 使用Key Selector Functions // some ordinary POJO public class WC {public String word; public int count;} DataSet<WC> words = // [...] DataSet<WC> wordCounts = words .groupBy( new KeySelector<WC, String>() { public String getKey(WC wc) { return wc.word; } }) .reduce(/*do something*/); Passing Functions to Flink 1. 实现function interface class MyMapFunction implements MapFunction<String, Integer> { public Integer map(String value) { return Integer.parseInt(value); } }); data.map (new MyMapFunction()); 或使用匿名类， data.map(new MapFunction<String, Integer> () { public Integer map(String value) { return Integer.parseInt(value); } }); 2. 使用Rich functions Rich functions provide, in addition to the user-defined function (map, reduce, etc), four methods:open,close,getRuntimeContext,andsetRuntimeContext. These are useful for parameterizing the function (seePassing Parameters to Functions), creating and finalizing local state, accessing broadcast variables (seeBroadcast Variables, and for accessing runtime information such as accumulators and counters (seeAccumulators and Counters, and information on iterations (seeIterations). Rich functions的使用和普通的function是一样的，不同的就是，多4个接口函数，可以用于一些特殊的场景，比如给function传参，或访问broadcast变量，accumulators和counter，因为这些场景你需要先getRuntimeContext class MyMapFunction extends RichMapFunction<String, Integer> { public Integer map(String value) { return Integer.parseInt(value); } }); Execution Configuration ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment(); ExecutionConfig executionConfig = env.getConfig(); enableClosureCleaner()/disableClosureCleaner(). The closure cleaner is enabled by default. The closure cleaner removes unneeded references to the surrounding class of anonymous functions inside Flink programs. With the closure cleaner disabled, it might happen that an anonymous user function is referencing the surrounding class, which is usually not Serializable. This will lead to exceptions by the serializer. 对于Java，比如传入function也要生成function对象，这样里面的function是会reference这个对象的，其实这种情况，你需要的只是function逻辑，所以closureCleaner会去掉这个reference 这样的好处是，传输类对象时候，是要求对象可序列化的，如果每个去实现序列号接口很麻烦，不实现又会报错，所以这里干脆clean掉这个reference getParallelism()/setParallelism(int parallelism)Set the default parallelism for the job. 设置Job的全局的parallelism getExecutionRetryDelay()/setExecutionRetryDelay(long executionRetryDelay)Sets the delay in milliseconds that the system waits after a job has failed, before re-executing it. The delay starts after all tasks have been successfully been stopped on the TaskManagers, and once the delay is past, the tasks are re-started. This parameter is useful to delay re-execution in order to let certain time-out related failures surface fully (like broken connections that have not fully timed out), before attempting a re-execution and immediately failing again due to the same problem. This parameter only has an effect if the number of execution re-tries is one or more.getExecutionMode()/setExecutionMode(). The default execution mode is PIPELINED. Sets the execution mode to execute the program. The execution mode defines whether data exchanges are performed in a batch or on a pipelined manner. 和失败重试相关的配置 enableObjectReuse()/disableObjectReuse()By default, objects are not reused in Flink. Enabling theobject reuse modewill instruct the runtime to reuse user objects for better performance. Keep in mind that this can lead to bugs when the user-code function of an operation is not aware of this behavior. 这个由于Java什么都要生成对象，比如function，所以会生成大量重复对象，这个可以打开object重用，提高性能 enableSysoutLogging()/disableSysoutLogging()JobManager status updates are printed toSystem.outby default. This setting allows to disable this behavior. 打开和关闭系统日志 getGlobalJobParameters()/setGlobalJobParameters()This method allows users to set custom objects as a global configuration for the job. Since theExecutionConfigis accessible in all user defined functions, this is an easy method for making configuration globally available in a job. 可以设置Job全局参数 其他的参数都是序列化相关的，不列了 Data Sinks Data sinks consume DataSets and are used to store or return them. Data sink operations are described using anOutputFormat. 可以custom output format: 比如写数据库， DataSet<Tuple3<String, Integer, Double>> myResult = [...] // write Tuple DataSet to a relational database myResult.output( // build and configure OutputFormat JDBCOutputFormat.buildJDBCOutputFormat() .setDrivername("org.apache.derby.jdbc.EmbeddedDriver") .setDBUrl("jdbc:derby:memory:persons") .setQuery("insert into persons (name, age, height) values (?,?,?)") .finish() ); 还有个功能，可以做locally的排序， DataSet<Tuple3<Integer, String, Double>> tData = // [...] DataSet<Tuple2<BookPojo, Double>> pData = // [...] DataSet<String> sData = // [...] // sort output on String field in ascending order tData.print().sortLocalOutput(1, Order.ASCENDING); // sort output on Double field in descending and Integer field in ascending order tData.print().sortLocalOutput(2, Order.DESCENDING).sortLocalOutput(0, Order.ASCENDING); Debugging 本地执行，LocalEnvironement final ExecutionEnvironment env = ExecutionEnvironment.createLocalEnvironment(); DataSet<String> lines = env.readTextFile(pathToTextFile); // build your program env.execute(); 便于调式的datasouce， final ExecutionEnvironment env = ExecutionEnvironment.createLocalEnvironment(); // Create a DataSet from a list of elements DataSet<Integer> myInts = env.fromElements(1, 2, 3, 4, 5); // Create a DataSet from any Java collection List<Tuple2<String, Integer>> data = ... DataSet<Tuple2<String, Integer>> myTuples = env.fromCollection(data); // Create a DataSet from an Iterator Iterator<Long> longIt = ... DataSet<Long> myLongs = env.fromCollection(longIt, Long.class); 便于输出的datasink， DataSet<Tuple2<String, Integer>> myResult = ... List<Tuple2<String, Integer>> outData = new ArrayList<Tuple2<String, Integer>>(); myResult.output(new LocalCollectionOutputFormat(outData)); Iteration Operators Iterations implement loops in Flink programs. The iteration operators encapsulate a part of the program and execute it repeatedly, feeding back the result of one iteration (the partial solution) into the next iteration. There are two types of iterations in Flink:BulkIterationandDeltaIteration. 参考，https://ci.apache.org/projects/flink/flink-docs-release-0.10/apis/iterations.html BulkIteration就是正常的Iteration，每次都处理全量数据 DeltaIteration，就是每次都只处理部分数据并delta更新，效率更高 Semantic Annotations Semantic annotations can be used to give Flink hints about the behavior of a function. 目的是做性能优化，优化器在明确知道function读参数的使用情况，比如如果知道某些field只是做forward，就可以保留它的sorting or partitioning信息 有3种语义annotation， Forwarded Fields Annotation 表示，输入的某个field会原封不动的copy到输出的某个field 下面的例子，表示输入的第一个field会copy到输出的第3个field 可以看到，输出tuple的第三个field是val.f0 @ForwardedFields("f0->f2") public class MyMap implements MapFunction<Tuple2<Integer, Integer>, Tuple3<String, Integer, Integer>> { @Override public Tuple3<String, Integer, Integer> map(Tuple2<Integer, Integer> val) { return new Tuple3<String, Integer, Integer>("foo", val.f1 / 2, val.f0); } } Non-Forwarded Fields 和上面相反，除指定的fields，其他fields都是原位置copy 例子，除输入的第二个field，其他都是原位置copy @NonForwardedFields("f1") // second field is not forwarded public class MyMap implements MapFunction<Tuple2<Integer, Integer>, Tuple2<Integer, Integer>> { @Override public Tuple2<Integer, Integer> map(Tuple2<Integer, Integer> val) { return new Tuple2<Integer, Integer>(val.f0, val.f1 / 2); } } Read Fields 表明这个fields会在function被读到或用到， 表明，输入的第一个field和第4个field会被读到或用到 @ReadFields("f0; f3") // f0 and f3 are read and evaluated by the function. public class MyMap implements MapFunction<Tuple4<Integer, Integer, Integer, Integer>, Tuple2<Integer, Integer>> { @Override public Tuple2<Integer, Integer> map(Tuple4<Integer, Integer, Integer, Integer> val) { if(val.f0 == 42) { return new Tuple2<Integer, Integer>(val.f0, val.f1); } else { return new Tuple2<Integer, Integer>(val.f3+10, val.f1); } } } Broadcast Variables Broadcast variables allow you to make a data set available to all parallel instances of an operation, in addition to the regular input of the operation. This is useful for auxiliary data sets, or data-dependent parameterization. The data set will then be accessible at the operator as a Collection. Broadcast: broadcast sets are registered by name viawithBroadcastSet(DataSet, String), and Access: accessible viagetRuntimeContext().getBroadcastVariable(String)at the target operator. // 1. The DataSet to be broadcasted DataSet<Integer> toBroadcast = env.fromElements(1, 2, 3); DataSet<String> data = env.fromElements("a", "b"); data.map(new RichMapFunction<String, String>() { @Override public void open(Configuration parameters) throws Exception { // 3. Access the broadcasted DataSet as a Collection Collection<Integer> broadcastSet = getRuntimeContext().getBroadcastVariable("broadcastSetName"); } @Override public String map(String value) throws Exception { ... } }).withBroadcastSet(toBroadcast, "broadcastSetName"); // 2. Broadcast the DataSet 这个场景，就是有些不大的公共数据，是要被所有的实例访问到的，比如一些查询表 上面的例子，会将toBroadcast设置为广播变量broadcastSetName，这样在运行时，可以用getRuntimeContext().getBroadcastVariable获取该变量使用 Passing Parameters to Functions 应该是如果将参数传递给function类，这个完全由java冗余导致的 首先，当然可以用类构造函数来传参数， ataSet<Integer> toFilter = env.fromElements(1, 2, 3); toFilter.filter(new MyFilter(2)); private static class MyFilter implements FilterFunction<Integer> { private final int limit; public MyFilter(int limit) { this.limit = limit; } @Override public boolean filter(Integer value) throws Exception { return value > limit; } } 自定义MyFilter，构造函数可以传入limit 也可以使用withParameters(Configuration) DataSet<Integer> toFilter = env.fromElements(1, 2, 3); Configuration config = new Configuration(); config.setInteger("limit", 2); toFilter.filter(new RichFilterFunction<Integer>() { private int limit; @Override public void open(Configuration parameters) throws Exception { limit = parameters.getInteger("limit", 0); } @Override public boolean filter(Integer value) throws Exception { return value > limit; } }).withParameters(config); 可以用withParameters将定义好的config传入function 然后用RichFunction的Open接口，将参数解析出来使用 这样和上面的比有啥好处，我怎么觉得上面那个看着更方便些？可以用匿名类? 当然你也可以用全局参数，这个和广播变量有什么区别？相同点就是都是全局可见，全局参数只能用于参数形式，广播变量可以是任意dataset Setting a custom global configuration Configuration conf = new Configuration(); conf.setString("mykey","myvalue"); final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment(); env.getConfig().setGlobalJobParameters(conf); Accessing values from the global configuration public static final class Tokenizer extends RichFlatMapFunction<String, Tuple2<String, Integer>> { private String mykey; @Override public void open(Configuration parameters) throws Exception { super.open(parameters); ExecutionConfig.GlobalJobParameters globalParams = getRuntimeContext().getExecutionConfig().getGlobalJobParameters(); Configuration globConf = (Configuration) globalParams; mykey = globConf.getString("mykey", null); } // ... more here ... Accumulators & Counters 用于分布式计数，job结束的时候，会全部汇总 Flink currently has the followingbuilt-in accumulators. Each of them implements theAccumulatorinterface. IntCounter,LongCounterandDoubleCounter: See below for an example using a counter. Histogram: A histogram implementation for a discrete number of bins. Internally it is just a map from Integer to Integer. You can use this to compute distributions of values, e.g. the distribution of words-per-line for a word count program. //定义和注册counter private IntCounter numLines = new IntCounter(); getRuntimeContext().addAccumulator("num-lines", this.numLines); //在任意地方进行计数 this.numLines.add(1); //最终取得结果 myJobExecutionResult.getAccumulatorResult("num-lines") Execution Plans 首先可以打印出执行plan，json格式， final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment(); ... System.out.println(env.getExecutionPlan()); 打开这个网页， The HTML document containing the visualizer is located undertools/planVisualizer.html. 将Json贴入，就可以看到执行计划， Web Interface Flink offers a web interface for submitting and executing jobs. If you choose to use this interface to submit your packaged program, you have the option to also see the plan visualization. The script to start the webinterface is located underbin/start-webclient.sh. After starting the webclient (per default onport 8080), your program can be uploaded and will be added to the list of available programs on the left side of the interface. 也可以通过web interface来提交job和查看执行计划
2017-04-06
优秀的个人博客,低调大师

Application Architecture Guide 2.0 chapter 1

第一章程序架构基础 目标 程序架构基础 标准关键程序建构术语和原则 促使现代架构形成的关键因素 概要 定义程序架构就是定义结构化的解决方案的过程，这个解决方案可以满足所有的技术和操作的需求，以及最佳化的一般质量特性，如性能，安全和可管理性。 这包括广泛因素的一系列的决策，每个决策都考虑到了对质量，性能，可维护性和程序的成功。下面我们就展开对这一过程的描述，并使用其中包含的信息，你可以构建一个包含所有要点的架构。并可以使这个架构部署在你选择的基础之上，并且符合所有的原始的目标和需求。本章是实践性的程序架构的基础。首先从高层介绍架构和设计，之后详细介绍具体的架构和设计。提示符和以上采用同样方式。最后，本章会给出关键术语和原则。理解这些会帮助你从本指南中获得最大的收益，并成为一个更好的架构师。通过学习本章，你会从整体角度理解结束和架构及设计，主要的你必须考虑的因素和你在设计架构是可以使用的主要方法。 这会帮助你理解一些架构和程序风格，质量特性和交叉相关的概念以及分析和展示架构。 什么是软件架构？ 软件架构一般被定义为一个系统的机构和各种结构的集合。 一些我们熟知的工业专家在架构相关决策的基础上扩展了这个定义。 Kruchten, Booch, Bittner,和Reitman对架构的定义 根据工作的经验Philippe Kruchten, Grady Booch, Kurt Bittner和Rich Reitman继承并提炼了架构的定义。他们的定义是：“软件架构包括了一些列的显著的软件系统组织的决策，包括： ü选择结构元素和他们组成系统的接口。 ü明确这些元素之间的协同行为 ü把这些结构的和行为的元素组成一个更大的子系统 ü指导这种组织的架构风格 软件架构同时包括功能性、可用性、可靠性、性能、复用性、广泛性以及经济和技术约束、权衡和美观等相关问题。” Fowler给出的架构定义 在《企业应用架构模式》一书中Martin Fowler在解释架构的时候，指出了一些重复发生的主题： •从最高层分解系统。 •决策很难修改。 •一个系统中存在多个架构。 •架构会影响一个系统的生命期。 •最后，架构按照系统的各个总要部分分解。 Bass, Clements, Kazman对架构的定义 在《软件架构实践（第二版）》中Bass, Clements,和Kazman这样定义软件架构： “一个软件或者是一个计算系统的架构就是这个系统的结构或者结构的集合，包括软件的各种元素，这些元素的外部可见的属性和他们之间的关系。架构主要是关于公共接口的，这些元素的内部细节—需要在内部实现的细节与架构无关。” 我们为什么需要架构设计？ 想其他的复杂结构一样，软件必须建立在一个稳定的基础之上。没有考虑到关键的用例，没有处理好常见的问题，或者没有对长期的关键决策做好评估，会导致一个应用的失败。先待工具和平台会帮助你简化创建应用的难度，但是他们不会在需求之上建立你的应用程序。不好的架构导致的风险包括不稳定，不支持业务需求，甚至软件部署到工作环境时无法工作。考虑软件架构时，从高层考虑一下问题： •软件如何部署到生产环境？ •用户会如何使用软件？ •质量需求有哪些，比如安全，性能，并发，国际化以及配置？ 架构和设计 当前或者在你部署了你的软件后，会影响到你的架构的趋势是什么？Martin Fowler说道：“专家一级的开发人员对于项目有个共同的理解。这个共同的理解叫做‘架构’。这个理解包括如何把系统分块以及这些不同的模块如何通过接口相互作用。这些模块通常又由更小的模块组成，但是架构只包括这些可以呗所有开发人员理解的模块以及他们之间的接口。”。所以结构主要集中在如何分解出模块和他们之间的接口。选择数据机构和算法实现已经不是架构了。不要用硬性快速的法则去区分架构和设计的区别，把他们两者结合起来才更有意义。在某些场合下，决策本质上来说更接近与架构。在其他情况下，决策又更接近于设计，并帮助你实现架构。 用户，业务和系统目标 系统架构的建立需要考虑到用户，系统和业务目标。每个领域都需要列出关键场景，重要的质量属性（比如，可维护性），关键客户要求实现的功能。如果可以，开发时，考虑每个领域中的成功的度量标准。 图1 在每个领域中都会有权衡，并且平衡点也必须找到。比如，反应时间可能是一个用户体验方面的要求但是系统管理者却不愿意投资到硬件上。而平衡点可能只是满足用户80%反应时间要求。 架构的目标 通过理解用例，软件架构可以建立业务需求和技术需求之间的桥梁，并在软件中实现用例。架构的目的就是为了辨别出对软件的结构有重大影响的需求。好的架构可以有效的降低业务的变化对解决方案的影响。一个好的设计可以非常灵活的适应由于时间引起的硬件和软件技术的变化，以及用户场景和需求的变化。一个架构师必须考虑设计决策，质量属性（比如：性能和安全）固有平衡，用户、系统和业务需求等对整体的影响。 架构应该包括的一下的内容： •找出系统的结构但是隐藏实现的细节。 •实现全部的场景用例。 •列出所有的利益相关者关注的问题。 • 实现架构的方法 功能和质量需求全都处理。任意的架构，实现架构的方式，都必须列出全部的关键决策。至少你必须确定你要建立的架构的类型，建立架构的风格并且处理这些问题的交叉部分。通过这个指南，我们会通过基本的架构来列出在你的架构中需要考虑的各个方面。基本架构在下图中。 图2 此外，对于基本架构来说，你可以用下列的方式来帮助你定义你自己的架构。第一步是你的要建立的应用的类型。接着，你必须理解你的软件如何部署。一旦你明白了你应用的类型以及该软件如何部署，你就可以开始用你认为合适的风格和技术来建立你的软件。最终，你需要把考虑质量属性和交叉点容纳到你的设计中。 应用类型 作为设计和组织软件架构过程的一部分，选择正确的应用类型极为关键。这取决于需求和基础结构的限制。本指南包括了一下的应用类型： l移动设备的移动应用 l主要运行在客户机上的富客户端应用 l部署在Internet上的支持丰富的UI和媒体的RIA应用 部署策略 当你设计架构时，你必须考虑到共同的方针和处理过程，以及你要部署你软件的基础。目标环境是灵活的还是不变的，你的软件必须考虑到在目标环境中存在的限制。你的应用程序必须同时考虑到质量属性比如安全和性能以及可维护性。有时你必须因为网络技术和协议做出必要的权衡。尽早在设计阶段分辨出需求和基础结构的限制和需求。这会帮助你选择一个正确的部署技术，帮助你尽早解决软件和基础架构之间的冲突。 架构风格 一种架构风格就是一些列的组成该风格的原则。每种风格都定义了一系列的规则，这些规则指明了你可以用于组成系统的组件，在软件组件之间关系，各个组件组织到一起的约束，以及他们组织到一起的方式。架构风格的例子是客户机/服务器，基于组件的，分层架构，消息总线，MVC，三层/N层，面向对象，以及面向服务的架构（SOA）。很多的因素会影响你对架构风格的选择。这些因素包括你所在开发单位的设计和实现的能力，开发人员的经验和能力，以及可以获得的架构约束和部署场景。 合适的技术 当为您的软件选择技术的时候，要考虑的关键因素是你所要开发的软件的类型，和软件部署技术以及架构风格。技术的选择也会受到代发单位策略，基础架构限制，技术资源等的影响。你必须比较在你的需求的基础上考虑你选择的技术，并且在做出任何的决定之前考虑其他的因素。 质量属性 质量属性可以把你的思考集中在几个设计中必须解决的关键的问题上。在你的需求的基础之上，你可能要考虑在本指南中列出的全部的质量属性，或者这些质量属性的一个子集。比如，每个软件都必须考虑安全和性能，但不是每个设计都需要考虑互操作性或者可升级性。首先理解你的需求和部署场景，这样哪些质量属性对你的设计很重要。记住质量属性会出现冲突。比如，安全性需要对性能和可用性做出权衡。在做安全性方面的设计时，分析并理解关键的权衡，就会避免边缘作用对软件产生显著的影响。在质量属性设计中可以考虑一下的指导： l质量是系统级属性，是和功能分离的。 l从技术的角度讲，质量属性的实现与否或实现如何可以区分一个软件的好坏。 l有两种质量属性：一种是在运行是度量的，一种只能通过审查来度量。 l分析质量属性之间的权衡。 当你考虑质量属性时，有几个问题你需要考虑： 什么是你软件要求的关键的质量属性？在设计的过程中找出他们。 什么是你软件要求的关键的质量属性？在设计的过程中找出他们。 客户的接手标准是什么，这表明了你是否符合需求。 关注交叉点 交叉点代表了软设计中与某一层无关的关键域。比如，你可能想在展示层和数据访问层缓存数据。你也需要设计一个在每一层都工作的异常处理框架。另外，你还设计每一层都可以使用的日志系统以及设计一个在不同层之间通讯的功能。权限管理也存在于不同的层次之间，所以你必须决定如何在系统中传递一个认证并使持有这个认证的用户可以访问特定的系统资源。一下列表描述了架构中必须考虑的交叉点： 身份认证：决定如何验证用户并在不同层之间传递认证的身份。 权限：在每一层和信任授权之间的授权正确。 缓存：确定什么是需要缓存的，缓存位置以改善软件的性能和响应时间。设计缓 存时一定要考虑到网页和网络应用的问题。 通讯：选择和是网络技术，减少网络调用并防止敏感数据在网络中传输。 异常管理：在边界捕获异常。不再终端用户钱显示敏感信息。 规范应用程序和日志：规范全部的业务和系统关键事件，并详细记录日志以便重建事件。不要记录敏感信息。 敏捷架构 敏捷架构假设软件的设计会随着时间的改变儿改变，并且你不知道为整个系统设计架构需要知道的所有信息。你的设计，总的来说，会在实施阶段获得了更多的信息后，或是在实际环境中测试过后加以改进。由于在设计的开始阶段无法全面的了解需求，而不断在头脑中改进架构。用敏捷的方法设计架构时，需要考虑一下问题： 在架构中，如果你出错了给整个系统带来最大风险的基本模块有那些。 架构中的那些部分是最容易改变的，或者说那部分的设计退后之后对系统的影响最小？ 你的关键假设是什么，如何测试他们？ 何种情况会使你重新考虑你的设计？ 不要过分设计，不要做无法验证的假设。相反的，保持对未来的变化开放的想法，而不是把自己逼到墙角。有些方面，如果重新设计会给你带来很高成本的话，是你在设计的初期就必须确定好的，尽快确认这些域，并认真分析他们。 敏捷架构设计的关键原则 敏捷架构包括一些关键的原则： 建模适应改变：任何地点只要可能，设计你的软件，以使之可以适应新的需求和 挑战。 建模分析、减低风险：给风险建模以理解风险和易出问题的地方。 建模和试图只是沟通和协调的工具：设计原则行业设计变更的沟通是敏捷设计的 关键所在。用户建模和其他的可视化手段可以使得沟通更加有效并可以使得 团队针对设计的变更做出迅速的应对。 确认关键建模决策：用本指导的框架来理解工程决策，区分出经常容易出错的地方。最后可以第一时间找出这些关键决策，以使设计更加灵活和更容易适应变化。 增量和迭代架构方式 敏捷架构通过增量和迭代的方式达到改进的目的。就是不要在第一次就把搜有的事情都作对。尽可能的设计，之后在需求和假设的基础上验证你的设计。迭代的在你的设计中添加细节，以使你在第一次就在重要问题上做出正确的决策，之后关注细节。一个常犯的错误就是过早的进入到了细节问题，并在错误的假设上做出错误的决策，或者无法评价架构是否有效。保证架构的基本方面是正确的，并测试这个架构，同时考虑一下问题： •在这个架构中我做了什么样的假设？ •这个架构能满足什么样的显式或隐性的需求？ •这样架构设计方式会有什么样的风险？ •通过何种方法应对关键的危机？ •以何种方式改进架构的基本设计或最近一个版本的架构？ 基本架构和待测架构 一个基本架构就是对现存系统的描述—也就是你系统当前的结构。如果是一个全新的架构，那么你的基本架构就是这个架构设计的最高层描述，待测架构也将从这里产生。一个待测架构包括应用类型，部署架构，架构风格，技术选择，质量属性和交叉点。 Architectural Spikes An architectural spike是一个软件的每个小块的端到端的测试。目的就是减低风险并测试潜在路径。只要你评价你的架构，你就会用这些spike来发现不同的场景，但这不会对软件现有的设计产生任何影响。An architectural spike可以得出一个待测架构，这个待测架构可以在基础架构的基础上进行测试。如果一个待测架构是一个改进，那么它就可以作为下一个待测架构的基础架构。这样的迭代和增量方式使你可以在第一时间摆脱较大的风险。用迭代的方式得出架构，并用架构测试证明每个基本架构都是一个改进。考虑下面的问题，以帮助你测试一个新的待测架构： •这个架构会导致新的风险吗？ •这个架构会降低已知的风险吗？ •这个架构满足其他的新增的需求吗？ •这个架构支持架构级的用例吗？ •这个架构列出了质量属性了吗？ •这个架构列出了增加的交叉点了吗？ 架构级的用例 架构级用例的符合以下标准： •他们对成功和客户对软件部署的接受至关重要。 •他们可以从多方面考验设计，这对架构的评估至关重要。 当你确定好架构的架构级用例之后，你可以用他们来评估待测框架的成功与否。如果待测框架需要更多的用例，或者更加有效的评估架构，那么这个待测架构是上个基本架构的改进版本了。 分析和评估架构 用架构评价来确定你的基本架构和待测架构的灵活性。架构评价是成功的架构迭代中的关键部分。在架构评价中考虑以下几个技术： •架构级用例:用用例来测试设计对你软件的成功非常重要。这也是你设计中的重要一环。 •基于场景的评估：用场景来分析你的设计。并且要考虑到质量属性。基于场景的评估有： ü架构权衡分析方法。 ü软件架构分析方法。 ü和中间设计审查。 展示和沟通架构 把你的架构拿出来沟通在架构审查中非常重要，在架构的实施过程中尤其如此。最后，只要你的沟通质量好，那架构的质量就有了保障。你必须和不同的角色讨论你的架构，包括系统设计者、开发人员、系统管理员等。一个可以清晰展现架构图景的方法是决策图。决策图不是某个专业的东西而是一个可以帮助你和其他人沟通的抽象。 架构图景 理解当前让我们做出架构决策，和将来让我们做出改变的关键因素。这些关键因素受客户要求的影响，同时也受到业务要求的影响，比如更快的得到结果，更好的支持工作风格和工作流程的改变，以及更容易修改的设计。考虑一下关键趋势： •用户授权：一个支持用户授权的设计是比较灵活的，这样的设计是可配置的，并考虑到了用户使用体验。设计时考虑到用户的个性和可能的选择。允许用户选择软件如何对操作做出回应，而不是培训他们。理解关键的使用场景，并使软件尽量的简单易用，容易找到信息和使用。 •市场成熟：在现有平台和技术选择的基础上烤炉事成成熟度。在高层建立应用框架才有意义，这样唯一的确定什么在你的软件中是有价值的，而不是建立一些已经存在并可以复用的。评价模式可以提供很多的已经证明了的普遍问题的解决方案。 •灵活性和适应性：一个灵活的，适应性强的设计重点在与松耦合以支持复用。考虑插件式开发以支持扩展。考虑服务优先的技术比如SOA以支持互操作。 •未来趋势：当你构建你的软件的时候，理解未来可能在部署后对你的设计产生影响的趋势。比如，考虑富客户端和媒体，符合模块比如混搭应用，增加的网络贷款和可用性，移动终端的增加，硬件性能的持续改进，社区和个人出版的增加，云计算的和远程操作的增加等。 本文转自张昺华-sky博客园博客，原文链接：http://www.cnblogs.com/sunshine-anycall/archive/2009/01/17/1377377.html，如需转载请自行联系原作者
2017-11-07
优秀的个人博客,低调大师

Hadoop- The Definitive Guide 笔记二

TheApache Hadoop projectdevelops open-source software for reliable, scalable, distributed computing, including:Hadoop Core, our flagship sub-project, provides a distributed filesystem (HDFS) and support for the MapReduce distributed computing metaphor. HBasebuilds on Hadoop Core to provide a scalable, distributed database. Pigis a high-level data-flow language and execution framework for parallel computation. It is built on top of Hadoop Core. ZooKeeperis a highly available and reliable coordination system. Distributed applications use ZooKeeper to store and mediate updates for critical shared state. Hiveis a data warehouse infrastructure built on Hadoop Core that provides data summarization, adhoc querying and analysis of datasets. Pig Pig raises thelevel of abstractionfor processing large datasets. With MapReduce, there is a map function and there is a reduce function, and working out how to fit your data processing into this pattern, which often requires multiple MapReduce stages, can be a challenge. Pig is made up of two pieces: The language used to express data flows, calledPig Latin. The execution environment to run Pig Latin programs. There are currently two environments:local executionin a single JVM anddistributed executionon a Hadoop cluster. A Pig Latin program is made up of a series of operations, or transformations, that are applied to the input data to produce output. Taken as a whole, the operations describe adata flow, which the Pig execution environment translates into an executable representation and then runs.Under the covers, Pig turns the transformations into a series of MapReduce jobs, but as a programmer you are mostly unaware of this, which allows you to focus on the data rather than the nature of the execution. Pig is a scripting language for exploring large datasets. One criticism of MapReduce is that the development cycle is very long. Writing the mappers and reducers, compiling and packaging the code, submitting the job(s) and retrieving the results is a timeconsuming business, and even with Streaming, which removes the compile and package step, the experience is still involved.Pig’s sweet spot is its ability to process terabytes of data simply by issuing a half-dozen lines of Pig Latin from the console. 总结一下，Pig是什么，为什么需要这个project，那就要先从MapReduce模型的不足说起 实际使用MapReduce模型时，你首先会碰到的一个问题就是建模问题，给定一个复杂的数据分析问题，怎么样把它抽象，转化成一系列的MapReduce过程，这个往往是比较困难的，是有些技术含量的 在分析抽象成了一系列MapReduce过程，你还要用Java去实现他们，事实证明用Java去实现还是很麻烦的 那么有没有一种Framework，能够让用户仅仅focus在怎样处理data的data flow上，而把这些底层的MapReduce和Java给屏蔽掉，好，Pig粉墨登场了，这个是Yahoo捐给Apache的一个项目，也是模仿google的一个项目开发的。 Pig包含了Pig Latin，一种脚本语言，专门用于描述data flow，看底下的一个例子。 An Example Let’s look at a simple example by writing the program to calculate the maximum recorded temperature by year for the weather dataset in Pig Latin. The complete program is only a few lines long: -- max_temp.pig: Finds the maximum temperature by year records = LOAD ''input/ncdc/micro-tab/sample.txt'' AS (year:chararray, temperature:int, quality:int); filtered_records = FILTER records BY temperature != 9999 AND (quality == 0 OR quality == 1 OR quality == 4 OR quality == 5 OR quality == 9); grouped_records = GROUP filtered_records BY year; max_temp = FOREACH grouped_records GENERATE group, MAX(filtered_records.temperature); DUMP max_temp; 好，这样一段简单的代码就完成了前面也介绍过的，年最高温度的问题，可以和前面用java写的对比一下，你就再也不会想用java写code了 具体什么意思，下面分别解释， Grunt是Pig的命令行程序 grunt> records = LOAD ''input/ncdc/micro-tab/sample.txt'' >> AS (year:chararray, temperature:int, quality:int); For simplicity, the program assumes that the input is tab-delimited text, with each line having just year, temperature, and quality fields. Load这个文件，Load会默认文件中的每行包含一些由Tab隔开的field，每个field是什么意思，即元数据在AS里面指定 grunt> DUMP records; (1950,0,1) (1950,22,1) (1950,-11,1) (1949,111,1) (1949,78,1)grunt> DESCRIBE records; records: {year: chararray,temperature: int,quality: int} 这步你读过去就知道什么意思了grunt> filtered_records = FILTER records BY temperature != 9999 AND >> (quality == 0 OR quality == 1 OR quality == 4 OR quality == 5 OR quality == 9); grunt> DUMP filtered_records; (1950,0,1) (1950,22,1) (1950,-11,1) (1949,111,1) (1949,78,1) The third statement uses the GROUP function to group the records relation by the year field.grunt> grouped_records = GROUP filtered_records BY year; grunt> DUMP grouped_records; (1949,{(1949,111,1),(1949,78,1)}) (1950,{(1950,0,1),(1950,22,1),(1950,-11,1)}) grunt>max_temp = FOREACH grouped_records GENERATE group, MAX(filtered_records.temperature); grunt> DUMP max_temp; (1949,111) (1950,22) 想知道Pig怎么样parse这些脚本语言的，看看下面这段话 As a Pig Latin program is executed, each statement is parsed in turn. If there are syntax errors, or other (semantic) problems such as undefined aliases, the interpreter will halt and display an error message. The interpreter builds alogical planfor every relational operation, which forms the core of a Pig Latin program. The logical plan for the statement is added to the logical plan for the program so far, then the interpreter moves on to the next statement. It’s important to note that no data processing takes place while the logical plan of the program is being constructed. When the Pig Latin interpreter sees the first line containing the LOAD statement, it confirms that it is syntactically and semantically correct, and adds it to the logical plan, but it does not load the data from the file (or even check whether the file exists). Similarly, Pig validates the GROUP and FOREACH ... GENERATE statements, and adds them to the logical plan without executing them. The trigger for Pig to start processing is the DUMP statement (a STORE statement also triggers processing). At that point, the logical plan is compiled into aphysical planand executed. The type of physical plan that Pig prepares depends on the execution environment. For local execution, Pig will create a physical plan that runs in a single local JVM, whereas for execution on Hadoop, Pig compiles the logical plan into a series of MapReduce jobs. You can see the logical and physical plans created by Pig using theEXPLAINcommand on a relation (EXPLAIN max_temp; for example). In MapReduce mode, EXPLAIN will also show the MapReduce plan, which shows how the physical operators are grouped into MapReduce jobs. This is a good way to find out how many MapReduce jobs Pig will run for your query. 总结，Pig在Parse脚本的时候，不是一句一句执行的，而是一条一条去check正确性，都放到Logic Plan里面，再把Logic Plan转化为Physical Plan去执行，这里脚本是local还是distributed执行对于用户透明的，只在转化为Physical Plan的时候，系统做了不同的处理。 那么如果你想知道Pig将你的脚本转化为怎样的MapReduce过程，你可以通过Explain命令去查看， 这个很有意思。 下面列出了Pig Latin的关系操作，从这儿你大概可以看出Pig可以对数据做怎么样的操作Pig Latin relational operatorsCategory Operator Description Loading and storingLOAD Loads data from the filesystem or other storage into a relation STORE Saves a relation to the filesystem or other storage DUMP Prints a relation to the console FilteringFILTER Removes unwanted rows from a relation DISTINCT Removes duplicate rows from a relation FOREACH ... GENERATE Adds or removes fields from a relation STREAM Transforms a relation using an external program Grouping and joiningJOIN Joins two or more relations COGROUP Groups the data in two or more relations GROUP Groups the data in a single relation CROSS Creates the cross product of two or more relations SortingORDER Sorts a relation by one or more fields LIMIT Limits the size of a relation to a maximum number of tuples Combining and splittingUNION Combines two or more relations into one SPLIT Splits a relation into two or more relations HBase HBase is adistributed column-oriented databasebuilton top of HDFS. HBase is the Hadoop application to use when you requirereal-time read/write random-accessto very large datasets. 给出一个HBase的Usecase The canonical HBase use case is thewebtable, a table of crawled web pages and their attributes (such as language and MIME type) keyed by the web page URL. The webtable is large with row counts that run into the billions. Batch analytic and parsing MapReduce jobs are continuously run against the webtable deriving statistics and adding new columns of MIME type and parsed text content for later indexing by a search engine. Concurrently, the table is randomly accessed by crawlers running at various rates updating random rows while random web pages are served in real time as users click on a website’s cached-page feature. Concepts Whirlwind Tour of theData Model Applications store data intolabeled tables. Tables are made of rows and columns. Table cells—the intersection of row and column coordinates—areversioned. By default, their version is a timestamp auto-assigned by HBase at the time of cell insertion. A cell’s content is an uninterpreted array of bytes. Table row keys are alsobyte arrays, so theoretically anything can serve as a row key from strings to binary representations of longs or even serialized data structures.Table rows are sorted by row key, the table’s primary key. Row columns are grouped intocolumn families. All column family members have a common prefix, so, for example, the columns temperature:air and temperature: dew_point are both members of the temperature column family, whereas station:identifier belongs to the station family.The column family prefix must be composed of printable characters. The qualifying tail can be made of any arbitrary bytes. A table’s column families must be specified up front as part of the table schema definition, butnew column family members can be added on demand. Physically,all column family members are stored togetheron the filesystem. So, though earlier we described HBase as a column-oriented store, it would be more accurate if it were described as acolumn-family-oriented store. In synopsis, HBase tables are like those in an RDBMS, onlycells are versioned,rows are sorted, andcolumns can be added on the flyby the client as long as the column family they belong to preexists. 总结，从抽象上你可以理解为Hbase也是采用Table的结构，不同于严谨的关系表，它design的目的就是可扩展性，所以你不用定义每个 column的类型（都是byte arrays），column的个数也是可以每行都不一样的，不会象关系表为稀疏表占用大量的空间，所以他就是一种可扩展的比关系表更灵活的一种表结构。 可是你有没有想过，这种表结构为什么那么灵活，答案就是其实Table只是你想象出来的，它本身并不是什么真正的表结构，尽管Google称为BigTable. 参考：http://jimbojw.com/wiki/index.php?title=Understanding_Hbase_and_BigTable Fortunately,Google''s BigTable Paperclearly explains what BigTable actually is. Here is the first sentence of the "Data Model" section: A Bigtable is asparse,distributed,persistent multidimensional sorted map. The BigTable paper continues, explaining that: The map is indexed by arow key, column key, and a timestamp; each value in the map is anuninterpreted array of bytes. 是的，它只是sorted map by row key，对于它前面那些定语还是比较好理解的，可以参考原文。 Regions Tables are automaticallypartitioned horizontally by HBase into regions. Each region comprises a subset of a table’s rows. A region is defined by its first row, inclusive, and last row, exclusive, plus a randomly generated region identifier. HBase会自动为大表分region，这个对用户是透明的。Implementation Just as HDFS and MapReduce are built of clients, slaves and a coordinating master—namenode and datanodes in HDFS and jobtracker and tasktrackers in MapReduce—so is HBase characterized with anHBase master nodeorchestratinga cluster of one or more regionserver slaves. The HBase master is responsible for bootstrapping a virgin install, for assigning regions to registered regionservers, and for recovering regionserver failures. The master node is lightly loaded. The regionservers carry zero or more regions and field client read/write requests. They also manage region splits informing the HBase master about the new daughter regions for it to manage the offlining of parent region and assignment of the replacement daughters. HBase depends on ZooKeeper and by default it manages a ZooKeeper instance as the authority on cluster state. HBase Versus RDBMS HBase and other column-oriented databases are often compared to more traditional and popular relational databases or RDBMSs. As described previously, HBase is a distributed, column-oriented data storage system. It picks up where Hadoop left off by providing random reads and writes on top of HDFS. Strictly speaking, anRDBMSis a database thatfollows Codd’s 12 Rules. Typical RDBMSs arefixed-schema,row-orienteddatabases withACID propertiesand a sophisticated SQL query engine. The emphasis is onstrong consistency,referential integrity,abstraction from the physical layer, andcomplex queriesthrough the SQL language. You can easily create secondary indexes, perform complex inner and outer joins, count, sum, sort, group, and page your data across a number of tables, rows, and columns. For a majority of small- to medium-volume applications, there is no substitute for the ease of use, flexibility, maturity, and powerful feature set of available open source RDBMS solutions like MySQL and PostgreSQL. However, if you need to scale up in terms of dataset size, read/write concurrency, or both, you’ll soon find that the conveniences of an RDBMS come at an enormous performance penalty and make distribution inherently difficult. The scaling of an RDBMS usually involvesbreaking Codd’s rules, loosening ACID restrictions, forgetting conventional DBA wisdom, and on the way losing most of the desirable properties that made relational databases so convenient in the first place. Countless applications, businesses, and websites have successfully achieved scalable, fault-tolerant, and distributed data systems built on top of RDBMSs and are likely using many of the previous strategies. But what you end up with is something that isno longer a true RDBMS, sacrificing features and conveniences for compromises and complexities. HBase Enter HBase, which has the following characteristics:No real indexes Rows are stored sequentially, as are the columns within each row. Therefore, no issues with index bloat, and insert performance is independent of table size.Automatic partitioning As your tables grow, they will automatically be split into regions and distributed across all available nodes.Scale linearly and automatically with new nodes Add a node, point it to the existing cluster, and run the regionserver. Regions will automatically rebalance and load will spread evenly.Commodity hardware Clusters are built on1,000–1,000–5,000 nodes rather than $50,000 nodes. RDBMS are hungry I/O, which is the most costly type of hardware.Fault tolerance Lots of nodes means each is relatively insignificant. No need to worry about individual node downtime.Batch processing MapReduce integration allows fully parallel, distributed jobs against your data with locality awareness. If you stay up at night worrying about your database (uptime, scale, or speed), then you should seriously consider making a jump from the RDBMS world to HBase. 总结，关系数据库什么都好，成熟稳定，严谨，但就是没有办法处理large scale data， 现在有很多正对关系数据库的large scale 的优化的solution，不过那些都破坏了其本源的那些准则和属性，这样使得它很多优点无法体现。 所以面对Large Scale data，来试试HBase吧，上面列出了那么多优点，还犹豫什么^_^ 同类型的NoSql数据库 here are other projects competing for the same position in the stack, in particular Facebook''sCassandraand LinkedIn''sProject Voldemort ZooKeeper 分布式环境中大多数服务是允许部分失败，也允许数据不一致，但有些最基础的服务是需要高可靠性，高一致性的，这些服务是其他分布式服务运转的基础，比如naming service、分布式lock等，这些分布式的基础服务有以下要求： 高可用性 高一致性 高性能 对 于这种有些挑战CAP原则的服务该如何设计，是一个挑战，也是一个不错的研究课题，Apache的ZooKeeper也许给了我们一个不错的答案。 ZooKeeper是一个分布式的，开放源码的分布式应用程序协调服务，它暴露了一个简单的原语集，分布式应用程序可以基于它实现同步服务，配置维护和命 名服务等。（http://blog.csdn.net/cutesource/archive/2010/08/18/5822459.aspx） 对于Hadoop，它的namenode或jobtracker就是一种重要的基础的node，必须要保证他们的高可用性，这就可以用 zookeeper来维护他们。对于Hbase也是这样的，依赖zookeeper来协调MasterServer，和RegionServer。 那么ZooKeeper的工作原理是什么了 ZooKeeper is a highly available, high-performance coordination service. Data Model ZooKeeper maintains ahierarchical tree of nodescalledznodes. A znode stores data and has an associated ACL. ZooKeeper is designed for coordination (which typically uses small data files), not high-volume data storage, so there is a limit of 1 MB on the amount of data that may be stored in any znode. ZooKeeper是用来存放一些coordination信息的，这些信息一般比较小，在zookeeper中是以一种hierarchical tree的形式来存放，这些具体的信息就store在tree node中，称为znode。 Data access is atomic.所有读写操作都是原子的 Znodes are referenced by paths, which in ZooKeeper are represented as slash-delimited Unicode character strings, like filesystem paths in Unix. 比如/zoo/duck, /zoo/cow Ephemeral znodes Znodes can be one of two types:ephemeral or persistent. A znode’s type is set at creation time and may not be changed later. An ephemeral znode is deleted by ZooKeeper when the creating client’s session ends. Ephemeral znodes are ideal for building applications that needs to know when certain distributed resources are available. 这边znode是有一种叫零时znode，这种当client session结束时会自动delete，后面讲到的lock service就会用到 Sequence numbers Asequential znodeis given a sequence number by ZooKeeper as a part of its name. If a znode is created with the sequential flag set, then the value of a monotonically increasing counter (maintained by the parent znode) is appended to its name. Sequence numbers can be used toimpose a global ordering on eventsin a distributed system, and may be used by the client to infer the ordering. 后面讲到的lock service，you will learn how to use sequential znodes to build a shared lock Watches Watches allow clients to get notifications when a znode changes in some way. Watches are set by operations on the ZooKeeper service, and are triggered by other operations on the service. There is an example in “A Configuration Service” demonstrating how to use watches to update configuration across a cluster. Watch就是可以在znode对某种操作加上trigger，如exist，change，当这种操作发生时，watch就会发notification给client Operations Operation Description create Creates a znode (the parent znode must already exist) delete Deletes a znode (the znode may not have any children) exists Tests whether a znode exists and retrieves its metadata getACL, setACL Gets/sets the ACL for a znode getChildren Gets a list of the children of a znode getData, setData Gets/sets the data associated with a znode sync Synchronizes a client’s view of a znode with ZooKeeper 这儿列出对znode的操作 Implementation The ZooKeeper service can run in two modes. Instandalonemode, there is a single ZooKeeper server, which is useful for testing due to its simplicity (it can even be embedded in unit tests), but provides no guarantees of high-availability or resilience. In production, ZooKeeper runs inreplicatedmode, on a cluster of machines called anensemble. ZooKeeper achieves high-availability through replication, and can provide a service as long as a majority of the machines in the ensemble are up. 前面说了，Zookeeper通过hierarchical tree来保存信息，但是standalone模式，其实没有实用价值的，单点局限，一个挂了就挂了。所以只有replicated模式才是high- availability的，只要cluster中majority servers是正常的，那么Zookeeper服务就是可用的 但是有个问题， 你在多台server，及ensemble上保存data，怎样保证所有server上数据的一致性，即Consistency，他是用了如下的方法 ZooKeeper uses a protocol calledZabthat runs in two phases, which may be repeated indefinitely: Phase 1:Leader election The machines in an ensemble go through a process of electing a distinguished member, called theleader. The other machines are termedfollowers. This phase is finished once a majority (or quorum) of followers have synchronized their state with the leader. Phase 2:Atomic broadcastAll write requests are forwarded to the leader, which broadcasts the update to the followers.When a majority have persisted the change, the leader commits the update, and the client gets a response saying the update succeeded. The protocol for achieving consensus is designed to be atomic, so a change either succeeds or fails. It resembles two-phase commit. If the leader fails, the remaining machines hold another leader election and continue as before with the new leader. If the old leader later recovers, it then starts as a follower. Does ZooKeeper Use Paxos? No. ZooKeeper’s Zab protocol is not the same as the well-known Paxos algorithm Google’sChubbyLock Service , which shares similar goals with ZooKeeper, is based on Paxos. 但是可用认为Zookeeper是对Paxos的优化实现，对于Paxos的相关资料如下 Paxos Made Simple【翻译】http://blog.csdn.net/sparkliang/archive/2010/07/16/5740882.aspx Paxos在大型系统中常见的应用场景 http://timyang.net/distributed/paxos-scenarios/ Consistency The terms “leader” and “follower” for the machines in an ensemble are apt, for they make the point that a follower may lag the leader by a number of updates. This is a consequence of the fact that only a majority and not all of the ensemble needs to have persisted a change before it is committed. follower对数据的更新肯定会lag于leader，而leader当majority的follower persisted achange的时候就会commit Every update made to the znode tree is given aglobally unique identifier, called azxid(which stands for “ZooKeeper transaction ID”). Updates are ordered, so if zxid z1 is less than z2, then z1 happened before z2, according to ZooKeeper, which is the single authority on ordering in the distributed system. 只是为什么所有的更新都要发给leader，需要一个globlly id来保证update的时序性 The followingguarantees for data consistency flowfrom ZooKeeper’s design:Sequential consistency Updates from any particular client are applied in the order that they are sent.Atomicity Updates either succeed or fail. This means that if an update fails, no client will ever see it. Single system image A client will see the same view of the system regardless of the server it connects to.Durability Once an update has succeeded, it will persist and will not be undone. This means updates will survive server failures. 每个znode的更新都是先更新persist设备，即硬盘，再更新memory Timeliness The lag in any client’s view of the system is bounded, so it will not be out of date by more than some multiple of tens of seconds. This means that rather than allow a client to see data that is very stale, a server will shut down, forcing the client to switch to a more up-to-date server. 下面举两个利用zookeeper的例子吧 A Configuration Service One of the most basic services that a distributed application needs is a configuration service so that common pieces of configuration information can be shared by machines in a cluster. At the simplest level, ZooKeeper can act as a highly available store for configuration, allowing application participants to retrieve or update configuration files. Using ZooKeeper watches, it is possible to create an active configuration service, where interested clients are notified of changes in configuration. A Lock Service A distributed lock is a mechanism for providing mutual exclusion between a collection of processes. At any one time, only a single process may hold the lock. Distributed locks can be used for leader election in a large distributed system, where the leader is the process that holds the lock at any point in time. 这儿的leader election，不同于zookeeper的leader election， 这儿讲的是一种通用的算法。 The pseudocode for lock acquisition is as follows: 1. Create anephemeral sequential znodenamed lock- under the lock znode and remember its actual path name (the return value of the create operation). 2. Get the children of the lock znode and set a watch. 3. If the path name of the znode created in 1 has the lowest number of the children returned in 2, then the lock has been acquired. Exit. 4. Wait for the notification from the watch set in 2 and go to step 2. The idea is simple: first designate a lock znode, typically describing the entity being locked on, say /leader; then clients that want to acquire the lock create sequential ephemeral znodes as children of the lock znode. At any point in time, the client with the lowest sequence number holds the lock. 其实过程是这样的 当一个client需要aquire lock的时候，和zookeeper建立session，并创建一个ephemeral sequential znode，所以产生的znodename是按照这个client aquire时的情况递增的，比如前面已有client 产生过lock－1，这时候你去aquire就会产生lock－2 znode lock－number最小的那个znode所对应的client得到这个lock，当它用完这个lock，需要释放lock时，这需要断开这个 client session，因为创建的是ephemeral znode，所以当session断开的时候，znode会自动删除。 It will be notified that it has the lock by creating a watch that fires when znodes go away. 你可用看到， client设置的watch只能触发一次，所以当有个znode被删除的时候，会给所有的client发notification，client收到后 check删除的是不是前一个znode，如果是那么就得到了lock，如果不是还要继续设置watch，就是上面第4步 这个的触发模式，当client很多时，会比较低效，一下要发出大量的notification，而其中只有一条是有用的，所以应该优化成watch 某一个znode被删除的情况。 Server Monitor Imagine a group of servers that provide some service to clients. 必须保持一个group membership list用于用户查询那些server可用，并当server fail的时候将他从list里面删除，server recover后自动加到list中。 The membership list clearly cannot be stored on a single node in the network, as the failure of that node would mean the failure of the whole system (we would like the list to be highly available). Suppose for a moment that we had a robust way of storing the list. We would still have the problem of how to remove a server from the list if it failed. Some process needs to be responsible for removing failed servers, but note that it can’t be the servers themselves, since they are no longer running! 你看这个问题还是比较麻烦的， 首先不能存在单点，不然单点fail了，整个service都挂了， 那就是要存在多台服务器上，保持replica，那么多台服务器上的data consistency就是一个很大的问题。就算这个问题解决了， 我们怎么样监控这个server，并动态的把fail的从list中删除，我们可用单独的进程去做这事，但如果这个进程所在的server崩溃了，怎么 办，好，是不是已经头大了 OK，Zookeeper可以比较好的解决这个问题， 只是我的理解，书中没说 对于每个服务器，当它启动时，自动建立一个client和zookeeper建立session，并创建一个ephemeral znode，client会不断的发送heartbeat保持这个session 这样当所有server都启动时， 他们在zookeeper上都有一个代表自己的znode， 而zookeeper的这个hierarchical tree就构成了这个服务group membership list，那么zookeeper是replica的，不用担心单点问题 当某个服务器crash，那么它建立的client的session会结束，那么它创建的那个znode，会被自动删除，因为是ephemeral的znode 这样就不需要单独的进程去监控server情况，并特意把fail的server从list中删除 同样当server recover的时候，会再次自动的创建client 建立session，也不需要其他进程干涉 这样用户只要通过查询zookeeper就可用知道那些server是可用的了 艾，牛啊，这个zookeeper真是很不容易理解。。。 Hive 为什么需要Hive When we started using Hadoop, we very quickly became impressed by its scalability and availability. However, we were worried about widespread adoption, primarily because of the complexity involved in writing MapReduce programs in Java (as well as the cost of training users to write them). We were aware that a lot of engineers and analysts in the company understood SQL as a tool to query and analyze data and that a lot of them were proficient in a number of scripting languages like PHP and Python. As a result, it was imperative for us to develop software that couldbridge this gapbetween thelanguages that the users were proficientin and the languages required toprogram Hadoop. Hadoop在scalability和availability方面非常的好，但是对于用Java来编写Map Reduce程序比较麻烦也比较困难。大多数程序员对SQL，和类似PHP，Python的脚本语言比较熟悉，所以如果我们能够直接用SQL－like语 言来对HDFS中存放的海量数据进行查询和处理就会非常方便，那么Hive就可以提供这样的功能。 Hive产生的动机和Pig比较相似，都是为了开发一套基于Hadoop的统一编程接口，来降低开发和使用Map Reduce的门槛，他们的之间的是有一定的overlap的。不过Pig使用Pig latin脚本语言，而Hive使用SQL－like语言，Pig Latin is procedural, where SQL is declarative. 所以他们使用的usecase还是有所不同的，Pig Latin更适合用来描述这个Data flow的处理过程，而Hive适合用于对海量数据进行查询访问 http://developer.yahoo.com/blogs/hadoop/posts/2010/08/pig_and_hive_at_yahoo/ 对于两者的区别参看上面的这个链接，说的比较清楚 Let me begin with a little background on processing and using large data. Data processing often splits into three separate tasks: data collection, data preparation, and data presentation. Thedata preparationphase is often known as ETL (Extract Transform Load) or thedata factory. Thedata presentationphase is usually referred to as thedata warehouse. Pig(combined with a workflow system such as Oozie) is best suited for thedata factory, andHivefor thedata warehouse. Hive是什么 Hive is adata warehouse infrastructurebuilt on top of Hadoop and serves as the predominant tool that is used to query the data stored in Hadoop at Facebook. A system that couldmodel data as tablesandpartitionsand that could also provide aSQL-like languagefor query and analysis. Also essential was the ability to plug in customized MapReduce programs written in the programming language of the user’s choice into the query. 这儿说Hive是数据仓库，model data as tables，partitions，自然会想到Hive和Hbase有什么不同 http://stackoverflow.com/questions/24179/how-does-hive-compare-to-hbase Hive is an analytics tool. Just like pig, it was designed for ad hoc batch processing of potentially enourmous amounts of data by leveraging map reduce. Think terrabytes. Imagine trying to do that in a relational database... HBase is a column based key value store based on BigTable. You can''t do queries per say, though you can run map reduce jobs over HBase. It''s primary use case is fetching rows by key, or scanning ranges of rows. A major feature is being able to have data locality when scanning across ranges of row keys for a ''family'' of columns. 从上面这段可以看出，Hbase和Hive其实问题域是不一样的，Hbase主要是为Hadoop提供low latency的随机访问能力，而Hive是为Hadoop提供一套SQL－like的分析和查询工具，Hive并不能保证low latency。 Hive is based on Hadoop which is a batch processing system. Accordingly, this system does not andcannot promise low latencies on queries. For Hive queries response times for even the smallest jobs can be of the order of 5-10 minutes and for larger jobs this may even run into hours. From one perspective, Hive consists of five main components: a SQL-like grammar and parser, a query planner, a query execution engine, a metadata repository, and a columnar storage layout. Its primary focus is data warehouse-style analytical workloads, so low latency retrieval of values by key is not necessary. Data organization Data is organized consistently across all datasets and is stored compressed, partitioned, and sorted:Compression Almost all datasets are stored as sequence files using gzip codec. Older datasets are recompressed to use the bzip codec that gives substantially more compression than gzip. Bzip is slower than gzip, but older data is accessed much less frequently and this performance hit is well worth the savings in terms of disk space.Partitioning Most datasets are partitioned by date. Going forward, we are also going to be partitioning data on multiple attributes (for example, country and date).Sorting Each partition within a table is often sorted (and hash-partitioned) by unique ID (if one is present). 本文章摘自博客园，原文发布日期：2011-07-04
2017-05-01
优秀的个人博客,低调大师

OpenStack 系列之File Share Service(Manila) Install Guide

OpenStack File Share Service(Manila) For Centos Introduction： Manilais the File Shareservice project for OpenStack. To administer the OpenStack File Share service,it is helpful to understand a number of concepts like share networks, shares,multi-tenancy and back ends that can be configured with Manila. Whenconfiguring the File Share service, it is required to declare at least one backend. Manila can be configured to run in a single-node configuration or acrossmultiple nodes. Manila can be configured to provision shares from one or moreback ends. The OpenStack File Share service allows you to offer file-shareservices to users of an OpenStack installation. 1.Create the database. 1 2 3 4 mysql-uroot-p CREATEDATABASEmanila; GRANTALLPRIVILEGESONmanila.*TO 'manila' @ 'localhost' IDENTIFIEDBY 'password' ; GRANTALLPRIVILEGESONmanila.*TO 'manila' @ '%' IDENTIFIEDBY 'password' ; 2.Create users, roles，service and API endpoint. 1 2 3 4 5 6 7 8 9 openstackusercreate--password-promptmanila openstackroleadd--projectservice--usermanilaadmin openstackservicecreate--namemanila--description "OpenStackSharedFilesystems" share openstackendpointcreate\ --publicurlhttp: //x .x.x.x:8786 /v1/ %\(tenant_id\)s\ --internalurlhttp: //x .x.x.x:8786 /v1/ %\(tenant_id\)s\ --adminurlhttp: //x .x.x.x:8786 /v1/ %\(tenant_id\)s\ --regionRegionOne\ share 3.Install required packages on controller node.openstack-manila-api and openstack-manila-scheduler services will run on thecontroller node. 1 yum install openstack-manilapython-manilapython-manilaclient 4.Installed required packages on compute node.openstack-manila-share service will run on the compute node. 1 yum install openstack-mania-sharepython-manila 5.Edit the/etc/manila/manila.conffile and api-paste.ini 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 [DEFAULT] osapi_share_listen=0.0.0.0 api_paste_config= /etc/manila/api-paste .ini state_path= /var/lib/manila glance_host=X.X.X.X storage_availability_zone=nova rootwrap_config= /etc/manila/rootwrap .conf auth_strategy=keystone enabled_share_backends=backend1 nova_catalog_info=compute:nova:publicURL nova_catalog_admin_info=compute:nova:adminURL nova_api_insecure=False nova_admin_username=nova nova_admin_password=password nova_admin_tenant_name=service nova_admin_auth_url=http: //X .X.X.X:5000 /v2 .0 network_api_class=manila.network.neutron.neutron_network_plugin.NeutronNetworkPlugin debug=True verbose=True log_dir= /var/log/manila use_syslog=False rpc_backend=rabbit control_exchange=openstack amqp_durable_queues=False cinder_catalog_info=volume:cinder:publicURL neutron_api_insecure=False cinder_admin_username=cinder neutron_auth_strategy=keystone cinder_admin_password=password notification_driver=messaging neutron_admin_tenant_name=service cinder_cross_az_attach=True neutron_url=http: //X .X.X.X:9696 cinder_api_insecure=False cinder_admin_auth_url=http: //X .X.X.X:5000 /v2 .0 cinder_http_retries=3 cinder_admin_tenant_name=service neutron_admin_password=password neutron_admin_username=neutron neutron_admin_auth_url=http: //X .X.X.X:5000 /v2 .0 neutron_url_timeout=30 default_share_type=default sql_connection=mysql: //manila :password@x.x.x.x /manila [oslo_messaging_rabbit] rabbit_host=controller rabbit_userid=openstack rabbit_password=password [oslo_concurrency] lock_path= /var/lock/manila Warning!!! This Manila Share Backend is configure on Manila Share node,So Manila-api and Manila-scheduler not configure. 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 [backend1] service_network_division_mask=28 volume_name_template=manila-share-%s service_instance_network_helper_type=neutron max_time_to_build_instance=300 share_mount_path= /shares manila_service_keypair_name=manila-service service_network_name=manila_service_network interface_driver=manila.network.linux.interface.OVSInterfaceDriver service_network_cidr=10.254.0.0 /16 service_instance_flavor_id=100 service_instance_smb_config_path=$share_mount_path /smb .conf volume_snapshot_name_template=manila-snapshot-%s share_backend_name=backend1 smb_template_config_path=$state_path /smb .conf service_instance_name_template=manila_service_instance_%s driver_handles_share_servers=True service_instance_user=manila service_instance_password=manila service_image_name=manila-service-image path_to_private_key= /root/ . ssh /id_rsa path_to_public_key= /root/ . ssh /id_rsa .pub share_backend_name=GENERIC1 share_driver=manila.share.drivers.generic.GenericShareDriver share_helpers=CIFS=manila.share.drivers.generic.CIFSHelper,NFS=manila.share.drivers.generic.NFSHelper share_volume_fstype=ext4 max_time_to_attach=120 service_instance_security_group=manila-service connect_share_server_to_tenant_network=False Replace /etc/manila/api-paste .ini,ConfigureKeystoneauthtoken. service_protocol=http service_host=X.X.X.X service_port=5000 auth_host=X.X.X.X auth_port=35357 auth_protocol=http admin_tenant_name=service admin_user=manila admin_password=password signing_dir= /var/lib/manila auth_uri=http: //X .X.X.X:5000/ 6.Populate the Manila Database. 1 su -s /bin/sh -c "manila-managedbsync" manila 7.Start and enable manila services. 1 2 3 4 5 6 systemctlstartopenstack-manila-api systemctlstartopenstack-manila-scheduler systemctl enable openstack-manila-api systemctl enable openstack-manila-scheduler systemctlstartopenstack-manila-share systemctl enable openstack-manila-share 8.Upload manila service imageto glance. 1 2 wgethttps: //github .com /uglide/manila-image-elements/releases/download/0 .1.0 /manila-service-image .qcow2 glanceimage-create--name "manila-service-image-new" -- file manila-service-image.qcow2--disk- format qcow2--container- format bare--visibilitypublic--progress 9.Create default share type. 1 manila type -createdefaultTrue 10.Create Manila flavor type. 1 novaflavor-createmanila-service-flavor10012801 11.Create nova for manila Security Group. 1 2 3 4 5 6 7 8 9 novasecgroup-createmanila-service 'manila-servicedescription' novasecgroup-add-rulemanila-serviceicmp-1-10.0.0.0 /0 novasecgroup-add-rulemanila-servicetcp22220.0.0.0 /0 novasecgroup-add-rulemanila-servicetcp204920490.0.0.0 /0 novasecgroup-add-rulemanila-serviceudp204920490.0.0.0 /0 novasecgroup-add-rulemanila-serviceudp4454450.0.0.0 /0 novasecgroup-add-rulemanila-servicetcp4454450.0.0.0 /0 novasecgroup-add-rulemanila-servicetcp1371390.0.0.0 /0 novasecgroup-add-rulemanila-serviceudp1371390.0.0.0 /0 12.Create share network. 1 manilashare-network-create--namedevinshare--neutron-net- id XXXXXXX--neutron-subnet- id XXXXXXX 13.Create NFS share. 1 manilacreateNFS1--namedevin--share-networkdevinshare 14.Configure Manila access. 1 manilaaccess-allowdevinipXXX(instance_ip)--access-levelrw 本文转自Devin 51CTO博客，原文链接:http://blog.51cto.com/devingeng/1745324
2017-11-13
优秀的个人博客,低调大师

书籍:ASTQB-BCS移动测试基础指南 Mobile Testing An ASTQB-BCS Foundation Guide - ...

简介 移动测试是测试移动软件的功能，可用性和一致性的过程。 与标准软件测试类似，高效且有效的移动测试需要在软件测试人员通常需要的技能之上提供额外的技能。 有了这个必不可少的指南，符合ASTQB认证的移动测试人员课程大纲，您将获得开始成为熟练的移动测试人员所需的理解和技能。 参考资料 下载：https://www.jianshu.com/p/a252732f8f1c python测试开发项目实战-目录 本文涉及的python测试开发库 谢谢点赞！ 本文相关海量书籍下载 2018最佳人工智能机器学习工具书及下载(持续更新) Format Pdf Page Count 183 Pages 针对读者 软件测试、开发、产品等。
2019-04-15

资源下载

更多资源
Mario

Mario

马里奥是站在游戏界顶峰的超人气多面角色。马里奥靠吃蘑菇成长，特征是大鼻子、头戴帽子、身穿背带裤，还留着胡子。与他的双胞胎兄弟路易基一起，长年担任任天堂的招牌角色。

时间: 2026-04-11 大小: 211.28KB 下载: 56次
腾讯云软件源

腾讯云软件源

为解决软件依赖安装时官方源访问速度慢的问题，腾讯云为一些软件搭建了缓存服务。您可以通过使用腾讯云软件源站来提升依赖包的安装速度。为了方便用户自由搭建服务架构，目前腾讯云软件源站支持公网访问和内网访问。

时间: 2026-08-21 大小: 1MB 下载: 10次
Rocky Linux

Rocky Linux

Rocky Linux（中文名：洛基）是由Gregory Kurtzer于2020年12月发起的企业级Linux发行版，作为CentOS稳定版停止维护后与RHEL（Red Hat Enterprise Linux）完全兼容的开源替代方案，由社区拥有并管理，支持x86_64、aarch64等架构。其通过重新编译RHEL源代码提供长期稳定性，采用模块化包装和SELinux安全架构，默认包含GNOME桌面环境及XFS文件系统，支持十年生命周期更新。

时间: 2026-08-21 大小: 1.42GB 下载: 6次
Sublime Text

Sublime Text

Sublime Text具有漂亮的用户界面和强大的功能，例如代码缩略图，Python的插件，代码段等。还可自定义键绑定，菜单和工具栏。Sublime Text 的主要功能包括：拼写检查，书签，完整的 Python API ， Goto 功能，即时项目切换，多选择，多窗口等等。Sublime Text 是一个跨平台的编辑器，同时支持Windows、Linux、Mac OS X等操作系统。

时间: 2026-08-21 大小: 9.84MB 下载: 40次
热门标签
DeepSeek Jdk25 OpenAi HarmonyOS6 Nacos3 SpringBoot4 SpringCloud Docker Kubernetes Service Mesh Redis Gemini3 Rocky

欢迎您来访！

登录后，您将获得更多功能！
热门文章
联系我们

有任何问题或合作意向欢迎联系我们

Email: 99873273@qq.com

QQ: 99873273

用户登录
用户注册