Pramod Immaneni <pramod@datatorrent.com>Apache Apex PMC, Architect @DataTorrent IncMay 7th, 2016

The next generation native Hadoop platformIntroduction to Apache Apex

2

What is Apex• Platform and runtime engine that enables development of scalable and fault-tolerant distributed applications

• Hadoop native• Process streaming or batch big data• High throughput and low latency• Library of commonly needed business logic• Write any custom business logic in your application

3

Applications on Apex• Distributed processing

• Application logic broken into components called operators that run in a distributed fashion across your cluster

• Scalable• Operators can be scaled up or down at runtime according to the load and SLA

• Fault tolerant• Automatically recover from node outages without having to reprocess from

beginning• State is preserved• Long running applications

• Operators• Use library to build applications quickly• Write your own in Java using the API

• Operational insight – DataTorrent RTS• See how each operator is performing and even record data

4

Apex Stack Overview

5

Apex Operator Library - Malhar

6

Native Hadoop Integration

• YARN is the resource manager

• HDFS used for storing any persistent state

7

Application Development Model

A Stream is a sequence of data tuplesA typical Operator takes one or more input streams, performs computations & emits one or more output streams

• Each Operator is YOUR custom business logic in java, or built-in operator from our open source library• Operator has many instances that run in parallel and each instance is single-threaded

Directed Acyclic Graph (DAG) is made up of operators and streams

Directed Acyclic Graph (DAG)

Filtered

Stream

Output StreamTuple Tuple

Filtered Stream

Enriched Stream

Enriched

Stream

er

Operator

er

Operator

er

Operator

er

Operator

er

Operator

er

Operator

8

Advanced Windowing Support

Application window Sliding window and tumbling window

Checkpoint window No artificial latency

9

Application in Java

10

Operators

11

Operators (contd)

12

Partitioning and unificationNxM PartitionsUnifier

0 1 2 3

Logical DAG

0 1 2

1

1 Unifier

1

20

Logical Diagram

Physical Diagram with operator 1 with 3 partitions

0

Unifier

1a

1b

1c

2a

2b

Unifier 3

Physical DAG with (1a, 1b, 1c) and (2a, 2b): No bottleneck

Unifier

Unifier0

1a

1b

1c

2a

2b

Unifier 3

Physical DAG with (1a, 1b, 1c) and (2a, 2b): Bottleneck on intermediate Unifier

13

Advanced Partitioning

0

1a

1b

2 3 4Unifier

Physical DAG

0 4

3a2a1a

1b 2b 3b

Unifier

Physical DAG with Parallel Partition

Parallel Partition

Container

uopr

uopr1

uopr2

uopr3

uopr4

uopr1

uopr2

uopr3

uopr4

dopr

dopr

doprunifier

unifier

unifier

unifier

Container

Container

NIC

NIC

NIC

NIC

NIC

Container

NIC

Logical Plan

Execution Plan, for N = 4; M = 1

Execution Plan, for N = 4; M = 1, K = 2 with cascading unifiers

Cascading Unifiers

0 1 2 3 4

Logical DAG

14

Dynamic Partitioning

• Partitioning change while application is runningᵒ Change number of partitions at runtime based on statsᵒ Determine initial number of partitions dynamically

• Kafka operators scale according to number of kafka partitionsᵒ Supports re-distribution of state when number of partitions changeᵒ API for custom scaler or partitioner

2b

2c

3

2a

2d

1b

1a1a 2a

1b 2b

3

1a 2b

1b 2c 3b

2a

2d

3a

Unifiers not shown

15

How tuples are partitioned• Tuple hashcode and mask used to determine destination

partitionᵒ Mask picks the last n bits of the hashcode of the tupleᵒ hashcode method can be overridden

• StreamCodec can be used to specify custom hashcode for tuplesᵒ Can also be used for specifying custom serialization

tuple: {Name, 24204842, San Jose}

Hashcode: 001010100010101

Mask (0x11)

Partition

00 1

01 2

10 3

11 4

16

Custom partitioning• Custom distribution of tuples

ᵒ E.g.. Broadcast

tuple:{Name, 24204842, San Jose}

Hashcode: 001010100010101

Mask (0x00)

Partition

00 1

00 2

00 3

00 4

17

Fault Tolerance• Operator state is checkpointed to a persistent store

ᵒ Automatically performed by engine, no additional work needed by operator

ᵒ In case of failure operators are restarted from checkpoint stateᵒ Frequency configurable per operatorᵒ Asynchronous and distributed by defaultᵒ Default store is HDFS

• Automatic detection and recovery of failed operatorsᵒ Heartbeat mechanism

• Buffering mechanism to ensure replay of data from recovered point so that there is no loss of data

• Application master state checkpointed

18

Processing Guarantees - RecoveryAtleast once• On recovery data will be replayed from a previous checkpoint

ᵒ Messages will not be lostᵒ Default mechanism and is suitable for most applications

• Can be used in conjunction with following mechanisms to achieve exactly-once behavior in fault recovery scenariosᵒ Transactions with meta information, Rewinding output, Feedback

from external entity, Idempotent operationsAtmost once• On recovery the latest data is made available to operator

ᵒ Useful in use cases where some data loss is acceptable and latest data is sufficient

Exactly once• At least once + state recovery + operator logic to achieve

end-to-end exactly once

19

Stream Locality• By default operators are deployed in containers (processes)

randomly on different nodes across the Hadoop cluster

• Custom locality for streamsᵒ Rack local: Data does not traverse network switchesᵒ Node local: Data is passed via loopback interface and frees up

network bandwidthᵒ Container local: Messages are passed via in memory queues

between operators and does not require serializationᵒ Thread local: Messages are passed between operators in a same

thread equivalent to calling a subsequent function on the message

20

Monitoring ConsoleLogical View

21

Monitoring ConsolePhysical View

App Ideas

22

• Ingestion with some ETL• Social media trending• Dimensional Analytics on large data sets like weather data• Location tracking• Alerting with IoT streams• Free your imagination

Useful operator list

23

• LineByLineFileInputOperator, StringFileOutputOperator• JMSStringInputOperator, JMSStringSinglePortOutputOperator• KafkaSinglePortStringInputOperator, POJOKafkaOutputOperator• KinensisStringInputOperator, KinesisStringOutputOperator• JdbcPOJOInputOperator, JdbcPOJOOutputOperator• CsvParser• FilterOperator

Reference Resources

24

• Documentation – http://docs.datatorrent.com• Beginners guide - http://docs.datatorrent.com/beginner/

• JavaDoc - https://www.datatorrent.com/docs/apidocs/

Resources

25

• Apache Apex website - http://apex.apache.org/• Subscribe - http://apex.apache.org/community.html• Download - http://apex.apache.org/downloads.html• Twitter - @ApacheApex; Follow - https://twitter.com/apacheapex• Facebook - https://www.facebook.com/ApacheApex/• Meetup - http://www.meetup.com/topics/apache-apex• Free Enterprise License for Startups -

https://www.datatorrent.com/product/startup-accelerator/

We Are Hiring

26

• jobs@datatorrent.com• Developers/Architects• QA Automation Developers• Information Developers• Build and Release• Community Leaders

End

27