Download as PDF, PPTX
![Flink APIs 6 case class Word (word: String, frequency: Int) val lines: DataStream[String] = env.readFromKafka(...) lines.flatMap {line => line.split(" ").map(word => Word(word,1))} .keyBy("word”).timeWindow(Time.of(5,SECONDS)).sum("frequency") .print() val lines: DataSet[String] = env.readTextFile(...) lines.flatMap {line => line.split(" ").map(word => Word(word,1))} .groupBy("word").sum("frequency”) .print() DataSet API (batch): DataStream API (streaming):](https://image.slidesharecdn.com/neo4j-flink-meetup-160302233132/75/Batch-and-Stream-Graph-Processing-with-Apache-Flink-6-2048.jpg)
![Hello, Gelly! ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment(); DataSet<Edge<Long, NullValue>> edges = getEdgesDataSet(env); Graph<Long, Long, NullValue> graph = Graph.fromDataSet(edges, env); DataSet<Vertex<Long, Long>> verticesWithMinIds = graph.run( new ConnectedComponents(maxIterations)); val env = ExecutionEnvironment.getExecutionEnvironment val edges: DataSet[Edge[Long, NullValue]] = getEdgesDataSet(env) val graph = Graph.fromDataSet(edges, env) val components = graph.run(new ConnectedComponents(maxIterations)) Java Scala 11](https://image.slidesharecdn.com/neo4j-flink-meetup-160302233132/75/Batch-and-Stream-Graph-Processing-with-Apache-Flink-12-2048.jpg)
Uploaded byVasia Kalavri
Batch and Stream Graph Processing with Apache Flink
The document discusses batch and stream graph processing using Apache Flink, an open-source framework tailored for real-time data streaming and analysis. It highlights Flink's capabilities in handling both streaming and batch data through different APIs, introduces its graph processing library Gelly, and explains iterative graph processing and single-pass stream graph processing. Additionally, it covers key concepts like windowing, stateful operations, and challenges in maintaining graph structures during stream processing.
In this document
Powered by AI
Introduction to Apache Flink, an open-source, distributed data analysis framework with true streaming capabilities by Vasia Kalavri.
Overview of Apache Flink including its core features and APIs for batch and stream processing. Supports low latency and various data sources.
Integration of Apache Flink with various programming languages and environments (e.g., Java/Scala) to utilize its capabilities.
Discussion on the advantages of stream processing including lower latency and better management of data volumes.
Details on batch and streaming processing parameters, including data movement, stateful operations, and optimizations.
Example code demonstrating the usage of Flink APIs for processing data in both streaming and batch modes.
Explanation of windowing in Flink with examples of tumbling and sliding windows for data freshness.
Introduction of Flink's libraries (Gelly, Table, ML) for different data handling, especially graph processing.
Description of graph methods, properties, and iterative graph processing using Gelly with emphasis on transformations.
Iterative graph processing methodologies including scatter-gather and gather-sum-apply approaches with algorithm examples.
Introduction to Gelly-Stream for handling single-pass graph processing in dynamic environments.
Brief on static and dynamic graph processing within Flink's framework.
Overview of challenges faced in graph streaming such as maintaining structures and efficiency in updates.
Description of methods for handling graph streaming efficiently by maintaining graph summaries.
Discussion on batch connected components algorithm with iterative computation demonstrated step by step.
Introduction to stream connected components analysis using disjoint set structures.
Techniques for maintaining and updating graph components incrementally through stream processing.
Detailed discussion on Gelly-Stream for enriching the DataStream API with new abstractions for graph streams.
Exploration of operations available on GraphStreams including vertex and edge manipulations.
Real-time analysis of connected components in graphs through stream processing techniques.
Introduction to graph stream windowing for temporal aggregations and neighbor analysis.
Methodologies for finding similar nodes/vertices within a time-sliced graph structure.
Plans for future Gelly integration with projects like Neo4j and advancements in Flink graph processing capabilities.
Introduction to helpful resources for learning and exploring Gelly including guides and repositories.
Batch and Stream Graph Processing with Apache Flink
- 1. Batch & Stream GraphProcessing with Apache Flink Vasia Kalavri vasia@apache.org @vkalavri
- 2. Apache Flink • Anopen-source, distributed data analysis framework • True streaming at its core • Streaming & Batch API 2 Historic data Kafka, RabbitMQ, ... HDFS, JDBC, ... Event logs ETL, Graphs, Machine Learning Relational, … Low latency, windowing, aggregations, ...
- 3. Integration (picture notcomplete) POSIX Java/Scala Collections POSIX
- 4. Why Stream Processing? •Most problems have streaming nature • Stream processing gives lower latency • Data volumes more easily tamed 4 Event stream
- 5. Batch and Streaming Pipelinedand blocking operators Streaming Dataflow Runtime Batch Parameters DataSet DataStream Relational Optimizer Window Optimization Pipelined and windowed operators Schedule lazily Schedule eagerly Recompute whole operators Periodic checkpoints Streaming data movement Stateful operations DAG recovery Fully buffered streams DAG resource management Streaming Parameters
- 6. Flink APIs 6 case classWord (word: String, frequency: Int) val lines: DataStream[String] = env.readFromKafka(...) lines.flatMap {line => line.split(" ").map(word => Word(word,1))} .keyBy("word”).timeWindow(Time.of(5,SECONDS)).sum("frequency") .print() val lines: DataSet[String] = env.readTextFile(...) lines.flatMap {line => line.split(" ").map(word => Word(word,1))} .groupBy("word").sum("frequency”) .print() DataSet API (batch): DataStream API (streaming):
- 7. Working with Windows 7 Whywindows? We are often interested in fresh data!15 38 65 88 110 120 #sec 40 80 SUM #2 0 SUM #1 20 60 100 120 15 38 65 88 1) Tumbling windows myKeyStream.timeWindow( Time.of(60, TimeUnit.SECONDS)); #sec 40 80 SUM #3 SUM #2 0 SUM #1 20 60 100 15 38 38 65 65 88 myKeyStream.timeWindow( Time.of(60, TimeUnit.SECONDS), Time.of(20, TimeUnit.SECONDS)); 2) Sliding windows window buckets/panes
- 8. Working with Windows 7 Whywindows? We are often interested in fresh data! Highlight: Flink can form and trigger windows consistently under different notions of time and deal with late events! 15 38 65 88 110 120 #sec 40 80 SUM #2 0 SUM #1 20 60 100 120 15 38 65 88 1) Tumbling windows myKeyStream.timeWindow( Time.of(60, TimeUnit.SECONDS)); #sec 40 80 SUM #3 SUM #2 0 SUM #1 20 60 100 15 38 38 65 65 88 myKeyStream.timeWindow( Time.of(60, TimeUnit.SECONDS), Time.of(20, TimeUnit.SECONDS)); 2) Sliding windows window buckets/panes
- 9. Flink Stack Gelly Table ML SAMOA DataSet (Java/Scala)DataStream (Java/Scala) HadoopM/R Local Remote Yarn Embedded Dataflow Dataflow(WiP) Table Cascading Streaming dataflow runtime CEP 8
- 10.
- 11. Meet Gelly • Java& Scala Graph APIs on top of Flink • graph transformations and utilities • iterative graph processing • library of graph algorithms • Can be seamlessly mixed with the DataSet Flink API to easily implement applications that use both record-based and graph-based analysis 10
- 12. Hello, Gelly! ExecutionEnvironment env= ExecutionEnvironment.getExecutionEnvironment(); DataSet<Edge<Long, NullValue>> edges = getEdgesDataSet(env); Graph<Long, Long, NullValue> graph = Graph.fromDataSet(edges, env); DataSet<Vertex<Long, Long>> verticesWithMinIds = graph.run( new ConnectedComponents(maxIterations)); val env = ExecutionEnvironment.getExecutionEnvironment val edges: DataSet[Edge[Long, NullValue]] = getEdgesDataSet(env) val graph = Graph.fromDataSet(edges, env) val components = graph.run(new ConnectedComponents(maxIterations)) Java Scala 11
- 13. Graph Methods Graph Properties getVertexIds getEdgeIds numberOfVertices numberOfEdges getDegrees ... 12 Transformations map,filter, join subgraph, union, difference reverse, undirected getTriplets Mutations add vertex/edge remove vertex/edge
- 14. Neighborhood Methods graph.reduceOnNeighbors(new MinValue,EdgeDirection.OUT) 13
- 15. Iterative Graph Processing •Gelly offers iterative graph processing abstractions on top of Flink’s Delta iterations • Based on the BSP, vertex-centric model • scatter-gather • gather-sum-apply • vertex-centric (pregel)* • partition-centric* 14
- 16. Scatter-Gather Iterations • MessagingFunction: generatemessage for other vertices • VertexUpdateFunction: update vertex value based on received messages 15 Scatter Gather
- 17. Gather-Sum-Apply Iterations • Gather:compute one value per edge • Sum: combine the partial values of Gather to a single value • Apply: update the vertex value, based on the Sum and the current value 16 Gather ApplySum
- 18. Library of Algorithms •PageRank* • Single Source Shortest Paths* • Label Propagation • Weakly Connected Components* • Community Detection • Triangle Count & Enumeration • Graph Summarization • val ranks = inputGraph.run(new PageRank(0.85, 20)) • *: both scatter-gather and GSA implementations 17
- 19.
- 20. Real Graphs aredynamic Graphs are created from events happening in real-time 19
- 21.
- 22.
- 23.
- 24.
- 25.
- 26.
- 27.
- 28.
- 29.
- 30. Gelly on Streams 21 DataStreamDataSet DistributedDataflow Deployment DataStream
- 31. Gelly on Streams 21 DataStreamDataSet DistributedDataflow Deployment Gelly • Static Graphs • Multi-Pass Algorithms • Full Computations DataStream
- 32. Gelly on Streams 21 DataStreamDataSet DistributedDataflow Deployment Gelly Gelly-Stream • Static Graphs • Multi-Pass Algorithms • Full Computations DataStream
- 33. Gelly on Streams 21 DataStreamDataSet DistributedDataflow Deployment Gelly Gelly-Stream • Static Graphs • Multi-Pass Algorithms • Full Computations • Dynamic Graphs • Single-Pass Algorithms • Approximate Computations DataStream
- 34. Batch vs. StreamGraph Processing 22 Batch Stream Input Graph static dynamic Analysis on a snapshot continuous Response after job completion immediately
- 35. Graph Streaming Challenges •Maintain the graph structure • How to apply state updates efficiently? • Result updates • Re-run the analysis for each event? • Design an incremental algorithm? • Run separate instances on multiple snapshots? • Computation on most recent events only 23
- 36. Single-Pass Graph Streaming •Each event is an edge addition • Maintain only a graph summary • Recent events are grouped in graph windows 24
- 37. Graph Summaries • spannersfor distance estimation • sparsifiers for cut estimation • sketches for homomorphic properties graph summary algorithm algorithm~R1 R2 25
- 38.
- 39. Batch Connected Components •State: the graph and a component ID per vertex (initially equal to vertex ID) • Iterative Computation: For each vertex: • choose the min of neighbors’ component IDs and own component ID as new ID • if component ID changed since last iteration, notify neighbors 27
- 40.
- 41.
- 42.
- 43.
- 44. Stream Connected Components •State: a disjoint set data structure for the components • Computation: For each edge • if seen for the 1st time, create a component with ID the min of the vertex IDs • if in different components, merge them and update the component ID to the min of the component IDs • if only one of the endpoints belongs to a component, add the other one to the same component 32
- 45.
- 46. 31 52 54 76 86 42 ComponentID Vertices 1 1,3 1 43 2 5 6 7 8 34
- 47. 31 52 54 76 86 42 ComponentID Vertices 43 2 2,5 1 1, 3 1 43 2 5 6 7 8 35
- 48. 31 52 54 76 86 42 43 87 ComponentID Vertices 2 2,4, 5 1 1, 3 1 43 2 5 6 7 8 36
- 49. 31 52 54 76 86 42 43 87 41 ComponentID Vertices 2 2,4, 5 1 1, 3 6 6, 7 1 43 2 5 6 7 8 37
- 50. 52 54 76 86 42 43 87 41 ComponentID Vertices 2 2,4, 5 1 1, 3 6 6, 7, 8 1 43 2 5 6 7 8 38
- 51. 54 76 86 42 43 87 41 ComponentID Vertices 22, 4, 5 1 1, 3 6 6, 7, 8 1 43 2 5 6 7 8 39
- 52. 76 86 42 43 87 41 ComponentID Vertices 2 2,4, 5 1 1, 3 6 6, 7, 8 1 43 2 5 6 7 8 40
- 53. 76 86 42 43 87 41 ComponentID Vertices 6 6,7, 8 1 1, 2, 3, 4, 5 1 43 2 5 6 7 8 41
- 54. 86 42 43 87 41 ComponentID Vertices 6 6,7, 8 1 1, 2, 3, 4, 5 1 43 2 5 6 7 8 42
- 55. 42 43 87 41 ComponentID Vertices 6 6,7, 8 1 1, 2, 3, 4, 5 1 43 2 5 6 7 8 43
- 56.
- 57. API Requirements • Continuousaggregations on edge streams • Global graph aggregations • Support for windowing 45
- 58. Introducing Gelly-Stream 46 Gelly-Stream enrichesthe DataStream API with two new additional ADTs: • GraphStream: • A representation of a data stream of edges. • Edges can have state (e.g. weights). • Supports property streams, transformations and aggregations. • GraphWindow: • A “time-slice” of a graph stream. • It enables neighborhood aggregations
- 59. GraphStream Operations 47 .getEdges() .getVertices() .numberOfVertices() .numberOfEdges() .getDegrees() .inDegrees() .outDegrees() GraphStream ->DataStream .mapEdges(); .distinct(); .filterVertices(); .filterEdges(); .reverse(); .undirected(); .union(); GraphStream -> GraphStream Property Streams Transformations
- 60. Graph Stream Aggregations 48 result aggregate propertystreamgraph stream (window) fold combine fold reduce local summaries global summary edges agg global aggregates can be persistent or transient graphStream.aggregate( new MyGraphAggregation(window, fold, combine, transform))
- 61. Graph Stream Aggregations 49 result aggregate propertystream graph stream (window) fold combine transform fold reduce map local summaries global summary edges agg graphStream.aggregate( new MyGraphAggregation(window, fold, combine, transform))
- 62.
- 63.
- 64.
- 65.
- 66.
- 67.
- 68.
- 69.
- 70.
- 71.
- 72.
- 73.
- 74.
- 75.
- 76.
- 77.
- 78.
- 79.
- 80.
- 81.
- 82. Slicing Graph Streams 62 graphStream.slice(Time.of(1,MINUTE)); 11:40 11:41 11:42 11:43
- 83. Aggregating Slices 63 graphStream.slice(Time.of(1, MINUTE),direction) • Slicing collocates edges by vertex information • Neighborhood aggregations on sliced graphs source target
- 84. Aggregating Slices 63 graphStream.slice(Time.of(1, MINUTE),direction) • Slicing collocates edges by vertex information • Neighborhood aggregations on sliced graphs source target
- 85. Aggregating Slices 63 graphStream.slice(Time.of(1, MINUTE),direction) .reduceOnEdges(); .foldNeighbors(); .applyOnNeighbors(); • Slicing collocates edges by vertex information • Neighborhood aggregations on sliced graphs source target Aggregations
- 86.
- 87. Finding Matches Nearby 64 graphStream.filterVertices(GraphGeeks()) .slice(Time.of(15,MINUTE), EdgeDirection.IN) .applyOnNeighbors(FindPairs()) GraphStream :: graph geek check-ins wendy checked_in soap_bar steve checked_in soap_bar tom checked_in joe’s_grill sandra checked_in soap_bar rafa checked_in joe’s_grill
- 88. Finding Matches Nearby 64 graphStream.filterVertices(GraphGeeks()) .slice(Time.of(15,MINUTE), EdgeDirection.IN) .applyOnNeighbors(FindPairs()) slice GraphStream :: graph geek check-ins wendy checked_in soap_bar steve checked_in soap_bar tom checked_in joe’s_grill sandra checked_in soap_bar rafa checked_in joe’s_grill wendy steve sandra soap bar tom rafa joe’s grill GraphWindow :: user-place
- 89. Finding Matches Nearby 64 graphStream.filterVertices(GraphGeeks()) .slice(Time.of(15,MINUTE), EdgeDirection.IN) .applyOnNeighbors(FindPairs()) slice GraphStream :: graph geek check-ins wendy checked_in soap_bar steve checked_in soap_bar tom checked_in joe’s_grill sandra checked_in soap_bar rafa checked_in joe’s_grill wendy steve sandra soap bar tom rafa joe’s grill FindPairs {wendy, steve} {steve, sandra} {wendy, sandra} {tom, rafa} GraphWindow :: user-place
- 90. What’s next? • Integrationwith Neo4j (Input / Output) • OpenCypher on Flink/Gelly • Pregel and Partition-Centric Iterations • Integration with Graphalytics
- 91. Feeling Gelly? • GellyGuide https://ci.apache.org/projects/flink/flink-docs-master/libs/gelly_guide.html • Gelly-Stream Repository https://github.com/vasia/gelly-streaming • Gelly-Stream talk @FOSDEM16 https://fosdem.org/2016/schedule/event/graph_processing_apache_flink/ • An interesting read http://people.cs.umass.edu/~mcgregor/papers/13-graphsurvey.pdf • A cool thesis http://urn.kb.se/resolve?urn=urn:nbn:se:kth:diva-170425