Streams into Hadoop: Storm, Flume and Sqoop

After this lesson you will say why a Storm topology never finishes and which task each tuple reaches, build and size a Flume flow of sources, channels and sinks from its properties file, and copy a relational table into HDFS or Hive with Sqoop.

Page 943 shows a black-and-white photograph: a wooden chute on trestles through a forest, water rushing down it, a log riding the water. That is a flume — a channel that floats logs from where the trees are cut to the sawmill. Apache Flume took the name, and its design is in the picture: many places produce, one place consumes, the channel carries the load in between.

A web shop runs 64 web servers, each sending 2,000 log lines a second into Hadoop. One night HDFS — the sawmill — stops for an hour; the servers do not. Where do the 460 million lines of that hour wait, and how big must the waiting room be?

The answer is the one-line rule of thumb of page 963 — a single multiplication. On the way come Storm, which computes on data that never ends, and Sqoop, which does not stream at all.

A computation that never finishes

Every tool so far reads an input that has already ended. A MapReduce job (lesson 30), a Pig script or a Hive query (lesson 32) starts from files in HDFS, runs, and stops. Page 924: "Lots of tools for data (i.e. batch) processing", but "None of them are realtime systems which is becoming a real requirement for businesses". A fraud alert cannot wait for the file to end, because that file never ends.

The course notes describe the shift: in streaming, data "gets transformed and analyzed continuously as it is generated" instead of being stored first and processed later (notes, ch. 4). The input is a data stream, an unbounded sequence of time-varying data. The query stays registered and runs over the data flowing past, and since nobody can wait for the end of an unbounded input, it looks at limited windows of it — the clicks of the last minute. On the notes' throughput–latency map this is the "Continuous" zone: low latency, paid for with throughput.

Batch: the input has ended, and so does the job

Stream: the input never ends, nor does the topology

Fig. 1. Batch against stream. In batch, MapReduce, Pig and Hive read an input that is already complete, and the job stops when it has read it. In a stream, Storm's input is an "Unbounded sequence of tuples" (p. 927), and the topology keeps producing results until someone kills it.

Storm, page 924, was "Developed by BackType which was acquired by Twitter" and "provides realtime computation" with four promises: "Scalable" — each piece of the computation runs as many parallel tasks; "Guarantees no data loss"; "Extremely robust and fault-tolerant" — a dead worker is restarted, a dead machine's work reassigned; "Programming language agnostic". "Realtime" means low latency, results within seconds, not a hard deadline.

"Guarantees no data loss, so every tuple is processed exactly once." No: Storm's documentation promises at-least-once processing. A tuple that is not fully processed in time is replayed from the spout, so nothing is lost — but after a failure the same tuple can be processed twice, and a counter can count it twice.

Four words: stream, spout, bolt, topology

Stream — "Unbounded sequence of tuples" (p. 927). A tuple is a named list of values, such as (word: "fresh").

Spout — "Source of streams" (p. 928): it reads from outside — a message queue, a log, an API — and emits tuples.

Bolt — "Processes input streams and produces new streams: Can implement functions such as filters, aggregation, join, etc" (p. 929).

Topology — "Network of spouts and bolts" (p. 930): a directed graph in which each bolt subscribes to the streams of other components.

A topology is the streaming counterpart of a MapReduce job, with two differences: it is any graph, not a fixed map-then-reduce pair, and, in the words of the Storm tutorial, "a MapReduce job eventually finishes, whereas a topology processes messages forever (or until you kill it)". Storm is not the last stream processor of the course: Spark Streaming, in module 11 (lesson 38), accepts inputs from "Kafka, Flume, Twitter, TCP Sockets" (p. 1200) — Flume, the second tool of this lesson, among them.

The cluster

Page 925 draws three groups of machines (Fig. 2). Nimbus, the "Master node (similar to Hadoop JobTracker)", ships a topology's code to the cluster, assigns its tasks to machines and reassigns them when a machine fails — the JobTracker of lesson 31. The Supervisors "Run worker processes" on the worker machines. Between them sits ZooKeeper, "Used for cluster coordination".

Why a third party? Nimbus and the Supervisors never talk directly: assignments and heartbeats go through ZooKeeper. Both daemons are "fail-fast and stateless; all state is kept in Zookeeper or on a local disk" (Storm tutorial), so either can be killed and restarted without loss, and while Nimbus is down the running topologies keep running.

Fig. 2. The Storm cluster of p. 925. Nimbus, the master, wears the dashed outline of a coordinator. It never talks to the Supervisors directly: assignments and heartbeats pass through ZooKeeper, which keeps the state, so a crashed Nimbus does not stop the running topologies.

A topology is a graph of spouts and bolts that never finishes; Nimbus assigns its tasks, the Supervisors run them, ZooKeeper holds the state.

Which task gets the tuple?

"Spouts and bolts execute as many tasks across the cluster" (p. 931): every component gets a number of tasks, each running the same code on a share of the tuples. Scaling a topology means raising those numbers, not rewriting the code. The word count of the Storm tutorial — sentences in, a running count per word out — sets them in the last argument:

// Storm tutorial word count: 5 spout, 8 split, 12 count tasks
TopologyBuilder builder = new TopologyBuilder();
builder.setSpout("sentences", new RandomSentenceSpout(), 5);
builder.setBolt("split", new SplitSentence(), 8)
        .shuffleGrouping("sentences");
builder.setBolt("count", new WordCount(), 12)
        .fieldsGrouping("split", new Fields("word"));

Each emitted tuple raises the question of page 932: "When a tuple is emitted which task does it go to?" The answer is the stream grouping, declared on every edge when a bolt subscribes to a stream — .shuffleGrouping(…) and .fieldsGrouping(…) above. Page 933 lists four.

The four stream groupings of p. 933, and what each is for.

grouping	the slide	what it gives	use it for
shuffle	"pick a random task"	even load; any task may get any tuple	stateless steps: parse, filter, split a sentence
fields	"consistent hashing on a subset of tuple fields"	the same field values always reach the same task	state per key: a count per word, a session per user
all	"send to all tasks"	every task gets a copy	broadcast: a configuration change, a small lookup table
global	"pick task with lowest id"	one task sees the whole stream	a single total or top-N; no parallelism on that edge

Counting decides between them. A count task keeps a counter per word in its own memory. With shuffle grouping the "fresh" tuples spread over all the tasks, each holding a part of the count; no task knows the total. With fields grouping on "word", every "fresh" lands on one task, which therefore holds the true count — the streaming twin of the MapReduce rule "same key, same reducer" (lesson 30). Fig. 3 adds a report bolt on global grouping, so that one task sees every count.

Fig. 3. The word count of the Storm tutorial plus a report bolt, with 2, 3, 3 and 1 tasks (the circles). Every edge declares its grouping: any task may split any sentence, all tuples of one word must meet in one count task, and a single report task sees every count.

Now route the words yourself. The widget makes one insight impossible to miss: fields grouping keeps the same word on the same task — and that is what makes a parallel count correct.

The word count of Fig. 3 without the report bolt. The spout emits six short posts from a farmers' market, over and over; the split bolt turns each post into word tuples; each count task keeps its own counters. Set the number of tasks per component and the grouping of the edge split → count (the edge spout → split stays on shuffle), then emit sentences. The drawing shows the tuples of the last sentence, the table what every count task has counted so far. Fields grouping hashes the word (FNV-1a, a simple string hash) modulo the number of tasks; shuffle draws the task at random with a fixed seed, so every run after Reset is the same.

Shuffle grouping, 3 count tasks. Press One sentence: the post is "fresh eggs and fresh milk". Before you look: do its two "fresh" tuples reach the same count task? Then press Reset and Next round twice (12 sentences, 8 times "fresh"). Which task holds 8?
What you should see

The two "fresh" of the first post go to tasks 0 and 1. After 12 sentences no task holds 8: task 0 has 6, tasks 1 and 2 have 1 each. The sum is right, but no single task knows it, and the readout turns red.
Switch to fields grouping (the run starts again) and press Next round twice. Predict how many columns of the "fresh" row are filled.
What you should see

One: task 1 holds "fresh" 8 times, the true count, together with "milk" (4). Every word lives in exactly one task — the same word always goes to the same task, which is the insight of the widget. Look at the load too: task 0 counted only "honey" (6 tuples), task 2 counted 18. Fields grouping spreads the keys, not the tuples.
Choose all grouping and 4 count tasks, then press Next round twice: 36 word tuples. Predict the deliveries on the edge split → count, and what each task holds for "fresh".
What you should see

144 deliveries, 4 copies of every tuple, and every task holds "fresh" 8: exact, but counted four times over. All grouping is for things every task needs, such as a new configuration — not for sharing work.
Choose global grouping, still 4 count tasks, and run two rounds. Which task works? Then set the spout to 2 tasks and the split to 4 and run two rounds again — do the counts change?
What you should see

Task 0, the lowest id, receives all 36 tuples; tasks 1–3 stay idle, so the parallelism of the count bolt is wasted. Global grouping suits one final total or a top-N. More spout or split tasks change nothing in the counts: neither component keeps state, so any of their tasks can handle any sentence.

Flume: a pond at every station

Before anything can run on Hadoop, the data has to get there. Page 935 asks: "How do you get your data from these sources to Hadoop?" — the sources being an RDBMS and applications (Fig. 4). Every Hadoop component exposes a Java API (p. 936), so each application could write into HDFS itself; page 937 lists why that is not enough: "Multiple events", "Streaming", "Diverse sources", "Buffering" — "Flume and Sqoop can help!"

Fig. 4. The question of p. 935 — one arrow from two sources to Hadoop — answered with two tools. Events from applications are pushed into Hadoop continuously by Flume; the rows of a relational table are pulled in bulk by Sqoop, when an import runs.

The producers are many, their number "changes constantly", and they sit "in different governance zones, data centers, continents" (p. 940). Their data drifts too: a data center moves to IPv6, a developer "changes logs (again)", a JSON record gains a field (p. 941). Code on every producer would have to follow every change.

Page 939 gives the definition: "Apache Flume is a continuous data ingestion system" — open-source, reliable, scalable, manageable, customizable — that carries logs, click streams, sensor data, social streams and feeds to storage (HDFS, S3), to query systems (Hive, HBase, Cassandra…) or to search engines. It was "Originally designed to be a log aggregation system by Cloudera Engineers" and "Evolved to handle any type of streaming event data" (p. 943). Flume moves data; it does not compute on it.

The agent: source, channel, sink

The agent is the "Simplest unit in Flume" (p. 944): one process that connects "any number of sources to any number of data stores" through three kinds of component (p. 948).

Source — "Accepts incoming Data", "Scales as required", "Writes data to Channel".

Channel — "Stores data in the order received".

Sink — "Removes data from Channel", "Sends data to downstream Agent or Destination".

What travels is an event, "the base unit of communication between components" (p. 946): a header of metadata that "Can be used for routing content", and a body, the "actual content" (p. 947) — bytes that Flume never needs to understand. The arrows of page 946 matter: "Source pushes events", "Sink polls for events". The sink keeps asking the channel "Got any events?", and the channel never calls anyone. Page 966 adds the cardinalities: "Source – Channel: many-to-many", "Channel– Sink: one-to-many (sink reads only from one channel)".

Fig. 5. One agent, as on pp. 946–949. The source pushes events into the channel inside its transaction; the sink polls ("Got any events?"), takes a batch inside its own transaction and commits only once HDFS has it. The slides draw the channel as a database cylinder; it is a queue.

Transactions

Agents chain: page 945 draws a "Distributed Pipeline Architecture" of agents from the web logs to HDFS, each with its own channel, one agent's sink feeding the next agent's source. Page 949: "Source uses transactions to write to the channel", "Sink uses transactions to remove data from the channel", "Sink transaction commits only after successful transfer of data". The source commits a batch only when the channel has accepted all of it. The sink takes a batch, delivers it — to HDFS, or to the next agent's source, which commits it into its own channel — and only then commits, removing the batch. On any failure the transaction rolls back and the events stay put: a baton released only into another hand.

"This ensures no data loss in Flume pipeline"

The transactions make every hand-off safe while the agents run: an event leaves a channel only after the next hop has it. That is no loss in transit with a durable (file) channel, and it is at-least-once, not exactly-once: a batch retried after a failed commit can arrive twice. With a memory channel, the events in the channel are lost when the agent process dies.

Page 968 calls the memory channel "Not persistent", page 962 warns of the "duplication impact" of large batches, and the Flume user guide says that "any events still left in the memory channel when an agent process dies can't be recovered" — try the crash button of the widget below. Page 986 repeats the claim ("Flume never loses any data in transit between Agents"): the exam may print it, so know it and its conditions.

Why channels?

Page 951 gives three reasons: the channel "Buffers data and insulates downstream from load spikes", "Provides persistent store for data in case the process restarts", and "Provides flow ordering* and transactional guarantees". The asterisk has no footnote; page 986 qualifies it: a "weak ordering guarantee, i.e., in the absence of failures the data will arrive in the order it was received". The chart of page 951 tells a channel's life: nearly empty in steady state; growing when the destination fails; at "Channel Full" the source's puts fail too and the pressure moves upstream; once the destination is restored, the backlog drains.

Page 968 offers two kinds. A memory channel is an "In-memory queue", "Fast", "Not persistent", "Limited capacity". The other, "Disk", is the file channel of a configuration (type = file): it writes events to local disk, survives a restart and can be far larger, at the price of speed.

From three web servers to a converging flow

Pages 952–956 tell the case Flume was built for: web-server logs into HDFS. With "only 3 web servers at the time of launch", an "Ad-hoc solution will likely suffice!" (p. 953). Growth brings the challenges: "How do you manage your output paths on HDFS?" and "How do you maintain your client code in face of changing environment as well as requirements?"

One Flume agent between the servers and HDFS (p. 954) brings "Insulation from HDFS downtime", "Quick offload of logs from Web Server machines" and "Better Network utilization" — but "What if the Flume node goes down?" Two agents, every server connected to both (p. 955), give "Redundancy and Availability" and "Automatic load balancing and failover", until more servers arrive. The scalable answer is a converging flow (p. 956): agents near the servers, aggregated by tier 2 and tier 3 "before being put into destination system".

Before: no Flume (p. 953)

After: a converging flow (p. 956)

Fig. 6. The log aggregation case. Before: each web server writes to HDFS itself — fine at launch, but every server carries client code and manages its own output paths. After: tiers of agents converge on the destination; inner agents carry more traffic and deliver bigger batches, and the dotted paths to a second aggregator give failover. The slide draws 16, 4 and 3 agents in three tiers; the sketch has two tiers.

Why converge? HDFS prefers a few writers that write big files (lesson 31): every file costs memory on the NameNode. "Closer a tier is to the destination, larger the batch size it delivers downstream", hence the "Optimized handling of destination systems". Per agent, the event volume "is least in the outermost tier" and "highest in the innermost tier" (p. 958); per tier, "In steady state, all tiers carry same event volume", while "Transient variations in flow are absorbed and ironed out by channels" (p. 959). Same total, fewer agents: more for each.

Three rules of thumb

"What we know" (p. 960): the number of web servers, the log volume per server per unit time, the destination and its routing, the worst-case downtime of the destination. "What we will calculate": the number of tiers, the exit batch sizes, the channel capacity — one rule each, pp. 961–964:

\[ N_{t+1} = \left\lceil N_t / f \right\rceil, \quad 4 \le f \le 16 \] \[ B_t = V / N_t \] \[ C = r_{\text{in}}^{\max} \times T_{\text{out}}^{\max} \]

$N_t$: number of agents in tier $t$; tier 1 holds the client agents
$f$: fan-in: "One Aggregating Agent (A) can be used with anywhere from 4 to 16 client Agents" (p. 961)
$V$: steady-state event volume entering the flow, the same in every tier (p. 959)
$B_t$: exit batch size of one agent of tier $t$: "total exit data volume divided by number of Agents in a tier" (p. 962)
$C$: channel capacity: "worst case data ingest rate sustained over the worst case downstream outage interval" (p. 963)
$r_{\text{in}}^{\max}$: worst-case ingest rate, in events per second
$T_{\text{out}}^{\max}$: worst-case downstream outage, in seconds

The slides state the rules in words; the first formula is the 4-to-16 ratio applied tier after tier. The figure of p. 962 draws the second one: $n$ client agents send $k$ each, so one downstream agent must deliver $n * k + \Delta$ and two deliver $(n * k + \Delta)/2$ each, $\Delta$ being the extra room. Each rule has its "Considerations": the tiers must also provide "routing, load-balancing and failover"; "Having some extra room is good", and large batches make duplicates costly; "Multiple disks will yield better performance", and "Channel size impacts the back-pressure buildup in the pipeline" — a bigger channel delays the moment the producers are refused. Each has a "Test" too: load-test steady state, peak and fail-over, and expect to "need more disk space than the physical footprint of the data size". The web shop of the hook, step by step:

What we know: 64 web servers, 2,000 events per second each, 500 bytes per event; destination HDFS, no routing; worst outage of HDFS: one hour, 3,600 seconds.
Tiers (p. 961): one client agent per server gives $N_1 = 64$. A fan-in of 8 gives $N_2 = 8$. Another 8 would give a single aggregator, and a single agent is a single point of failure (p. 954); a fan-in of 4 gives $N_3 = 2$.
\[ 64 \to 8 \to 2 \]
Volume (p. 959): in steady state every tier carries what enters the flow — 64 MB per second at 500 bytes per event.
\[ V = 64 \times 2000 = 128\,000 \ \text{events/s} \]
Exit batch size per agent (p. 962): tier 1 sends 2,000 events/s per agent, tier 2 sends 16,000, tier 3 sends 64,000. Add head-room $\Delta$, and test fail-over: with one tier-3 agent down, the other must drain all 128,000 alone.
\[ B_3 = 128\,000 / 2 = 64\,000 \]
Channel capacity (p. 963): tier 3 must hold everything that arrives during the outage — 460.8 million events of 500 bytes, about 230 GB, so 115 GB per agent. That is a file channel spread over several disks; a memory channel cannot come close.
\[ \begin{aligned} C &= 128\,000 \times 3600 \\ &\approx 4.6 \times 10^{8} \ \text{events} \end{aligned} \]

An agent in ten lines of properties

Running Flume takes two steps (p. 970): "Configure agent(s) in property file", then "Run Flume agent from console". One file can describe several agents; --name picks the one to start.

$ flume-ng agent --conf-file spool-to-logger.properties \
  --name agent1

Page 971 is the whole file of a "1-1-1" agent — one source, one channel, one sink:

# spool-to-logger.properties (p. 971)
agent1.sources = source1
agent1.sinks = sink1
agent1.channels = channel1

agent1.sources.source1.channels = channel1
agent1.sinks.sink1.channel = channel1

agent1.sources.source1.type = spooldir
agent1.sources.source1.spoolDir = /Users/udi/tmp/spooldir

agent1.sinks.sink1.type = logger
agent1.channels.channel1.type = file

Read it in three blocks. First the names of the components of agent1. Then the wiring: a source lists its channels, plural, because it may feed several; a sink names its channel, singular — the rule of page 966 is in the key names. Last, each component's type and properties. This agent turns every line of every file dropped into the spooling directory into an event, keeps it in a file channel and prints it with the logger sink.

The catalogue (pp. 967–969). Sources: Spooling Directory — "Text Files, Immutable, Unique name", since the source tracks files by name and stops with an error if a file changes or a name comes back (Flume user guide) — SysLog, Exec (the output of a command such as tail -F) and Custom. Channels: Memory and Disk. Sinks: HDFS, HBase, "Console Log" and "Local Directory" — in a configuration, the logger sink (type = logger) and the file_roll sink.

The HDFS sink

# p. 972: the logger sink replaced by an HDFS sink
agent1.sinks.sink1.type = hdfs
agent1.sinks.sink1.hdfs.path = /tmp/flume
agent1.sinks.sink1.hdfs.filePrefix = events
agent1.sinks.sink1.hdfs.fileSuffix = .log
agent1.sinks.sink1.hdfs.inUsePrefix = _
agent1.sinks.sink1.hdfs.fileType = DataStream

hdfs is the short name of the class on the slide, org.apache.flume.sink.hdfs.HDFSEventSink. Files land in /tmp/flume as events.<timestamp>.log — the answer to page 953's "output paths". While a file is being written its name starts with _, and Hadoop jobs skip such files: no job reads half a file. The sink "rolls" — closes the file, opens a new one — and "By default, files are rolled over every 30 seconds", with rollCount and rollSize as further triggers (p. 973). fileType picks "SequenceFile = binary format", "DataStream = uncompressed files (text)" or "CompressedStream = compressed files" (p. 974).

The 30 seconds of p. 973 are hdfs.rollInterval. The Flume user guide gives the other two triggers defaults too — rollSize 1024 bytes, rollCount 10 events — and the first trigger to fire wins, so a busy sink with default settings closes a file after 10 events or 1 KB, whichever comes first, and fills HDFS with tiny files. Set a trigger to 0 to switch it off.

HTTP in, folders by header

Page 975 swaps the source for an HTTP server, in front of a memory channel of 1,000 events:

# p. 975: an HTTP source on port 8989 and a memory channel
httpagent.sources.http-source.type = org.apache.flume.source.http.HTTPSource
httpagent.sources.http-source.bind = localhost
httpagent.sources.http-source.port = 8989
httpagent.channels.ch.type = memory
httpagent.channels.ch.capacity = 1000

Clients post events in a "Fixed format: header + body" (p. 976): a JSON array of objects with headers and body. When the channel is full, the source answers HTTP 503 (Flume user guide): back-pressure the client can see.

$ curl -X POST -H 'Content-Type: application/json; charset=UTF-8' \
  -d '[{"headers": {"eventheader1": "event1", "eventheader2": "event2"},
        "body": "This is the body1"}]' \
  http://localhost:8989

"Bucketing" (p. 977) is the "Dynamic assignment of files to folders": %{topic} in the path is replaced, event by event, by the value of the header topic, so the slide's event with "topic" : "topic1" lands in /tmp/flume/http/topic1/. Header names are case-sensitive: the slide's "Topic" is topic in the path and in the event. Time-based bucketing (p. 978) uses date escapes:

# p. 977: one folder per value of the header topic
httpagent.sinks.hdfs-sink.hdfs.path = /tmp/flume/http/%{topic}

# p. 978: one folder per hour, read from the sink's own clock
agent1.sinks.sink1.hdfs.useLocalTimeStamp = true
agent1.sinks.sink1.hdfs.path = /tmp/flume/twitterinterceptor/%Y/%m/%d/%H

Time escapes need a timestamp header, which the HTTP and spooling sources do not add by themselves; useLocalTimeStamp = true lets the sink use its own clock instead. A folder per hour is the partition-per-directory layout that Hive reads (lesson 32).

The same events can also go to HBase (p. 979):

# p. 979: each event into the column test:pCol of the HBase table test_table
agent1.sinks.hbaseSink.type = org.apache.flume.sink.hbase.AsyncHBaseSink
agent1.sinks.hbaseSink.table = test_table
agent1.sinks.hbaseSink.columnFamily = test
agent1.sinks.hbaseSink.serializer = org.apache.flume.sink.hbase.SimpleAsyncHbaseEventSerializer
agent1.sinks.hbaseSink.serializer.payloadColumn = pCol

Several channels, selectors and interceptors

A source can "replicate or multiplex data across many channels", each drained by its own sink (p. 950). Page 980 wires one HTTP source to two channels — ch1 into an HDFS sink, ch2 into a logger sink:

# p. 980: one source, two channels, two sinks
httpagent.sources = http-source
httpagent.sinks = hdfs-sink log-sink
httpagent.channels = ch1 ch2

httpagent.sources.http-source.channels = ch1 ch2
httpagent.sinks.hdfs-sink.channel = ch1
httpagent.sinks.log-sink.channel = ch2

Which event goes where is decided by the source's channel selector (p. 981): "Replicating • Default • Every event sent everywhere", or "Multiplexing • routes events to channels based on their headers". Page 982 adds four lines — "If show = 1, route to ch1", if show = 2, to ch2. The selector belongs to the source: routing happens before the channels.

# p. 982: route by the header show (without these lines: replicating)
httpagent.sources.http-source.selector.type = multiplexing
httpagent.sources.http-source.selector.header = show
httpagent.sources.http-source.selector.mapping.1 = ch1
httpagent.sources.http-source.selector.mapping.2 = ch2

Social media is one more source (p. 983): a Twitter source takes the four credentials of a Twitter developer account, which the slide masks, and a list of keywords. Pages 985 and 978 complete its pipeline: keep the tweets about the election, file them into one HDFS folder per hour.

# p. 983: tweets as events (the slide masks the credentials as **)
agent1.sources.source1.type = org.apache.flume.source.twitter.TwitterSource
agent1.sources.source1.consumerKey = **
agent1.sources.source1.consumerSecret = **
agent1.sources.source1.accessToken = **
agent1.sources.source1.accessTokenSecret = **
agent1.sources.source1.keywords = @realDonaldTrump, @HillaryClinton

Last, an interceptor "can process events based on their contents or headers": "Simple data processing", "Overhead for Flume", "To be used with caution!" (p. 984). It runs inside the source on every event, so a slow one slows ingestion. Page 985 keeps only the events whose body matches a regular expression; excludeEvents = true would drop them instead.

# p. 985, with the registration line it needs: keep only events whose body contains "election"
agent1.sources.source1.interceptors = regexInterceptor
agent1.sources.source1.interceptors.regexInterceptor.type = regex_filter
agent1.sources.source1.interceptors.regexInterceptor.regex = .*election.*
agent1.sources.source1.interceptors.regexInterceptor.excludeEvents = false

"The three lines of page 985 install the filter." Not alone: an interceptor works only once its name is listed on the source, as in the first line above (a1.sources.r1.interceptors = i1 i2 in the Flume user guide). Without that line, nothing is filtered.

Now build flows and push load through them. The widget makes one insight visible: a channel absorbs the difference between what arrives and what can leave, and once it is full the refusal moves upstream — so its capacity must cover the worst ingest rate over the worst outage.

Pick a flow: one agent as on p. 975, one agent with two channels and two sinks as on p. 980, or four client agents converging on an aggregator. Four web servers post 100 events a second each; between t = 10 s and t = 30 s comes a load spike, an HDFS outage, or both. Channels are rows of cells that fill up. The properties file after the readouts is generated from the flow you chose. The rates (100, 300 and 600 events a second) and the file channel's 20,000 events are the widget's round numbers — a real file channel holds a million by default; the memory channel's 1,000 is the value of p. 975.

One agent, memory channel, HDFS down, spike × 1. Before you press Run: 400 events a second arrive and none can leave. When is the channel of 1,000 full, and how many events are refused by t = 30 s? Then repeat with the file channel.
What you should see

Full about 2.5 s into the outage (1,000 / 400); from then on the source refuses everything: about 7,000 events by t = 30 s. The rule of p. 963 asks for 400 × 20 = 8,000 events, as the last readout says. The file channel takes them all: it peaks near 8,000 at t = 30 s, then drains at 600 − 400 = 200 a second and is empty again around t = 70 s.
Converging flow, memory channels, HDFS down, spike × 1. Predict which channel fills first, and when the web servers start being refused.
What you should see

The aggregator, by t = 13 s. Its source then refuses the clients' batches (dashed arrows), and each client channel grows by 100 events a second: full by t = 23 s. Only then are the web servers refused — about 3,000 events by t = 30 s, against 7,000 with a single agent. Every hop buffers, and back-pressure travels upstream one tier at a time.
Converging flow, HDFS up (switch the outage off), spike × 3. Where is the spike absorbed — with memory channels, then with file channels?
What you should see

Each client gets 300 events a second and can send 300, so tier 1 passes the spike on. The aggregator gets 1,200 but writes only 600 to HDFS: with memory channels it is full by t = 12 s, the clients by t = 19 s, and about 7,000 events are refused. With file channels the aggregator absorbs (1,200 − 600) × 20 = 12,000 events and drains them over the next 60 s: "Transient variations in flow are absorbed and ironed out by channels" (p. 959).
Two channels, memory, HDFS down. With the replicating selector, does the logger keep printing during the outage? Switch to multiplexing and compare; read the selector lines in the file. Finally, press Crash during the outage, while the channels hold events — once with memory, once with file channels.
What you should see

Replicating puts every event into both channels, so once ch1 is full (t = 13 s) the source refuses everything and the logger starves too. With multiplexing only the show = 1 half is refused (about 3,000 events) and show = 2 keeps reaching the logger; the file gains the four selector lines of p. 982. A crash empties a memory channel — the events are lost, though the web servers were told they were delivered — while a file channel finds them on disk after the restart.

Page 986 sums Flume up: it "is suitable for large volume data collection, especially when data is being produced in multiple locations"; "Once planned and sized appropriately, Flume will practically run itself"; it gives a weak ordering guarantee, transactional exchange between agents, and "rich out-of-the box features such as contextual routing".

Sqoop: the nightly truck

The other source of page 935 is a relational database. Sqoop (p. 988): "Importing from relational DB sources", "PULL-based (no streams)", "Bulk imports", "With one command line action, import data from RDBMS to HDFS/Hive", with connectors for MySQL, Oracle, SQL Server, PostgreSQL. Its use cases (p. 989): an "Occasional copy for running map-reduce tasks", and "Keep RDBMS for transactional needs and periodically dump data on HDFS" — transactions stay on the database, analysis moves to Hadoop. Flume is a pipe that never closes; Sqoop is a truck that drives when called.

# p. 990: copy the table tradingDays of the MySQL database nseProd into HDFS, one mapper
sqoop import \
--connect jdbc:mysql://localhost:3306/nseProd \
--username=qt \
--password=password \
--table=tradingDays \
--target-dir /mysql/nseProd \
--m 1

Read it: the tool (import), the JDBC address of the database nseProd on a local MySQL, the credentials, the table, the HDFS directory to create, the number of mappers. Behind the command Sqoop runs a MapReduce job with map tasks only: each mapper reads its share of the table with its own SELECT and writes its own file (part-m-00000, …) into the target directory.

How the rows are split

The shares are ranges of a column, by default the primary key: Sqoop asks the database for its smallest and largest value and cuts that interval into as many ranges of equal width as there are mappers (Fig. 7). --split-by names another column — "explicitly specify the column to use for splitting rows between mappers" (p. 992) — when there is no primary key or it splits badly. Equal widths are not equal row counts: a key with big gaps leaves some mappers idle.

Fig. 7. A Sqoop import with four mappers. Sqoop asks for the smallest and largest value of the split column — the primary key by default — cuts that interval into four ranges of equal width, and gives each range to one mapper, which runs its own SELECT and writes its own file. The key values are invented for the picture.

"Sqoop uses the primary key to decide how many mappers to use, and for splitting the rows among mappers" · "--m 1 fixes 1 mapper / mandatory if the table you are importing doesn’t have a primary key, or if you are importing a query"

The number of mappers is set by -m (--num-mappers) and is four by default; the primary key only decides how the rows are split among them. One mapper is needed when the table has no primary key and no --split-by column is given. A query import can run in parallel too, with --split-by.

The Sqoop user guide: "By default, four tasks are used"; without a primary key and without --split-by "import will fail unless the number of mappers is explicitly set to one"; a free-form query is either run in parallel, for which "You must also select a splitting column with --split-by", or "imported serially, by specifying a single map task with -m 1". The exam may print the slide's sentence: know what it says and what Sqoop does.

Instead of a whole table, a query (p. 993): "Instead of full table", "$CONDITIONS mandatory". The slide prints only the option; the complete command needs a target directory and either one mapper or a split column:

# p. 993, completed: three columns of the 2016 rows, four mappers split on month
sqoop import \
--connect jdbc:mysql://localhost:3306/nseProd \
--username=qt \
--password=password \
--query 'select year, month, day from tradingDays where year=2016 and $CONDITIONS' \
--split-by month \
--target-dir /mysql/tradingDays2016 \
-m 4

Sqoop replaces the token $CONDITIONS in each mapper's copy of the query with that mapper's range of month. It is not a shell variable: the single quotes keep the shell away from it.

Into Hive, and only what is new

Two more options turn the import into a Hive table (p. 994). The target directory is new: like every MapReduce job, Sqoop refuses to write into an output directory that already exists, and /mysql/nseProd was taken by the first import.

# p. 994: the same table, registered in Hive as tradingDays
sqoop import \
--connect jdbc:mysql://localhost:3306/nseProd \
--username=qt \
--password=password \
--table=tradingDays \
--hive-import \
--hive-table=tradingDays \
--target-dir /mysql/table/tradingDays2 \
--m 1

Re-importing a whole table every night copies the same rows again and again. Page 995: "Import only the new data" — "Save an import configuration as a JOB and call it later on repetitively". The incremental options pick rows by a check column, "Ideally a timestamp of last update of the row": "LastValue saves the max value of the column at every import", and a "Further import will get data > LastValue" (p. 998). Mode lastmodified (p. 997) is for rows that get updated; the user guide's other mode, append, for tables that only grow with an increasing id. The saved job keeps the last value, so each sqoop job --exec myjob (p. 999) continues where the previous run stopped. The corrected job below carries one line the slide lacks, the flag --append — not to be confused with the mode append: every run writes into the same /mysql/nseProd, and without it Sqoop refuses to import into a directory that already exists (red pen below).

# p. 997, corrected: save an incremental import as the job myjob
sqoop job \
--create myjob \
-- import \
--connect jdbc:mysql://localhost:3306/nseProd \
--username=qt \
--password=password \
--table=tradingDays \
--target-dir /mysql/nseProd \
--append \
--m 1 \
--incremental lastmodified \
--check-column ts

# p. 999: run it now, and every night after
sqoop job --exec myjob

"--import \" … "--target-dir /mysql/nseProd \" / "--m 1" / "--incremental lastmodified" / "—check-column ts"

The tool to save comes after a bare -- and a space: -- import \. Every line but the last needs its \, so --m 1 \ and --incremental lastmodified \, and the last option is --check-column ts with two hyphens. Page 996 has the same --import. And a job meant to be called "repetitively" (p. 995) needs --append: /mysql/nseProd already exists — the import of p. 990 created it, and every run after the first finds it again.

The Sqoop user guide gives the syntax sqoop job (generic-args) (job-args) [-- [subtool-name] (subtool-args)] and the example sqoop job --create myjob -- import --connect …. As printed, the shell would end the command after --m 1. On the directory, the same guide: "If the destination directory already exists in HDFS, Sqoop will refuse to import and overwrite that directory's contents. If you use the --append argument, Sqoop will import data to a temporary directory and then rename the files into the normal target directory". The exam is matched against the slides: recognise the printed form, write the corrected one.

Two blind spots come with incremental imports. An updated row arrives in HDFS a second time, in a new file next to its old copy; Sqoop's merge tool, with --merge-key, folds the copies into one (Sqoop user guide). And a deleted row is never noticed: nothing changes in its check column.

Flume and Sqoop, the two answers to the question of p. 935.

	Flume	Sqoop
reads from	applications: logs, events, clicks, sensors, social streams	relational databases, through JDBC connectors
control	push: sources receive and push into channels (sinks poll)	pull: "PULL-based (no streams)"
timing	continuous, agents run all the time	occasional or periodic, when the command or job runs
unit	the event: headers + body, in transactional batches	rows of a table or of a query, in bulk
parallelism	more agents, tiers, several channels and sinks	`-m` mappers on ranges of the split column
writes to	HDFS (bucketed by header or time), HBase, logger, …	HDFS files or a Hive table

Back to the hook. The 460 million lines of that hour wait in the channels of a converging flow, the biggest ones in the tier next to HDFS: 128,000 events a second for 3,600 seconds (p. 963), about 230 GB — 115 GB for each of its two agents, on file channels over several disks, where the memory channel of p. 975 holds a thousand events. Counting what flows past is Storm's job, with fields grouping on the word; the product table that sits still in the shop's database comes by truck: Sqoop, once a night, --incremental lastmodified.

Exam-style questions

Storm, Flume and Sqoop are not among the practical topics of the rules, and no past paper has an exercise on them: they belong to the theory part. Below are theory questions in the forms the readme of the past papers names — multiple choice, true or false, open answers; wrong answers may carry a penalty, announced just before the exam — a computation with a rule the slides give, and one exercise in writing a configuration and a command.

A Storm topology counts the occurrences of each word in a stream of sentences. A bolt splits the sentences and emits one tuple per word to a counting bolt with several tasks, each keeping a counter per word. Which stream grouping lets the counting tasks share the work while the count of each word stays exact in a single task?

Shuffle grouping It shares the work, but the tuples of one word reach random tasks: each task holds a partial count, and no task holds the total.
All grouping Every task receives every tuple and counts everything: the counts are exact but computed in every task, so no work is shared.
Fields grouping on the field "word" The task is chosen by hashing the word ("consistent hashing on a subset of tuple fields", p. 933): the same word always reaches the same task, and different words spread over the tasks.
Global grouping All tuples go to the task with the lowest id: the counts are exact, but one task does all the work.

Show model answer

Fields grouping on "word". The tuple is routed by a hash of the chosen field, so all tuples with the same word reach the same task, which therefore holds the exact count for that word, while different words are spread over the tasks. Shuffle grouping would split each word's count over several tasks; all grouping would make every task count everything; global grouping would send everything to one task.

The point of the question is the link between grouping and state: a bolt that keeps state per key needs the tuples of that key in one place. Each wrong option is right about something else — shuffle for stateless bolts, all for broadcasts, global for one final total — which is what makes them tempting.

Describe the main concepts of Apache Storm — stream, spout, bolt, topology — and the components of a Storm cluster. In what does a Storm topology differ from a MapReduce job?

Show model answer

Storm is a distributed system for realtime computation, developed at BackType, which Twitter acquired. A stream is an unbounded sequence of tuples, a tuple being a named list of values. A spout is a source of streams: it reads data from an external source and emits tuples. A bolt processes one or more input streams and produces new streams; it can implement filters, aggregations, joins or any function. A topology is a network of spouts and bolts — a directed graph — and every component runs as many tasks across the cluster; a stream grouping (shuffle, fields, all, global) decides which task receives each tuple. The cluster has a master node, Nimbus, similar to Hadoop's JobTracker, which distributes the code, assigns tasks to machines and monitors failures; a ZooKeeper cluster used for coordination, which keeps the cluster state; and Supervisors on the worker machines, which run the worker processes. A MapReduce job reads a bounded input and finishes; a topology processes an unbounded stream with low latency and runs until it is killed.

A complete answer has four definitions, three components and the one difference "finishes / never finishes". If there is room, add why ZooKeeper sits between Nimbus and the Supervisors: they keep no state of their own, so either can be restarted, and the topologies keep running while Nimbus is down (Storm tutorial).

Explain how Apache Flume moves events from a source to a sink within an agent, and from one agent to the next, without losing them. What does this mechanism not guarantee?

Show model answer

A Flume agent hosts a source, a channel and a sink. The source accepts incoming data, turns it into events (headers with metadata, used for routing, and a body with the content) and writes them into the channel inside a transaction, which commits only when the channel has accepted the whole batch. The sink polls the channel and removes a batch inside its own transaction, and commits only after a successful transfer — when the destination, such as HDFS, has written it, or when the source of the next agent has committed it into its own channel. On any failure the transaction rolls back and the events stay in the channel to be retried. An event therefore leaves a channel only when the next hop holds it, and no event is lost in transit between agents. Not guaranteed: durability with a memory channel, which is not persistent — its events are lost if the agent process dies, so only a file (disk) channel survives a restart; exactly-once delivery, since a batch retried after a failed commit can arrive twice; and strict ordering, since Flume gives only a weak ordering guarantee: events arrive in the order received in the absence of failures.

The slides print "This ensures no data loss" (p. 949): repeat it, then add its conditions — the red pen of this lesson. The question asks for both halves: the mechanism first, then the three limits.

True or false: Apache Flume guarantees that events always reach the destination in the order in which the source received them.

True Only "in the absence of failures": page 986 calls it a "weak ordering guarantee". If an agent dies and its web servers fail over to the second agent (p. 955), the events held in the first agent's file channel arrive after newer ones.
False Page 986: "Flume provides weak ordering guarantee, i.e., in the absence of failures the data will arrive in the order it was received". "Always" is what makes the statement false.

Show model answer

False. Flume provides only a weak ordering guarantee: in the absence of failures, events arrive in the order in which they were received. After a failure — for example an agent that dies and whose producers fail over to another agent — events can arrive out of order.

Look for the absolute word: the slide's guarantee holds only without failures. Page 951's "flow ordering*" carries the same restriction in its asterisk.

Consider the farmers' market system of the Exam of 12 Sep 2025: farmers sell their products at local markets, and every sale is sent as an event to a tier of Flume agents that writes into HDFS. In the worst case the tier ingests 1,500 events per second, and HDFS may be unavailable for up to 40 minutes. Following the rule of thumb of the slides, what channel capacity does the tier need, in millions of events?

Show model answer

Channel capacity = worst-case ingest rate × worst-case downstream outage = 1,500 events/s × 2,400 s = 3,600,000 events, i.e. 3.6 million events, spread over the agents of the tier. At this size a file channel is needed, on several disks, with more disk space than the raw data.

The rule of p. 963, as worded on the summary of p. 964; the scenario is the paper's, the question is ours. The usual slip is to leave the minutes unconverted: 1,500 × 40 = 60,000 events cover only 40 seconds of outage.

Which of the following statements about Apache Flume and Apache Sqoop are true? Select all that apply.

In a Flume agent, a source may write to several channels, while a sink reads from exactly one channel. True: "Source – Channel: many-to-many", "Channel– Sink: one-to-many (sink reads only from one channel)" (p. 966).
The Flume channel pushes events to the sink as soon as they arrive. False: "Source pushes events", "Sink polls for events" (p. 946); the channel is passive.
With a memory channel, the events waiting in the channel survive a crash of the agent process. False: the memory channel is "Not persistent" (p. 968); only the file (disk) channel keeps events across a restart.
The multiplexing channel selector routes each event to channels according to the value of one of its headers. True: "routes events to channels based on their headers" (p. 981), e.g. header show = 1 to ch1 (p. 982).
Sqoop is pull-based: it reads the relational database in bulk when the import runs. True: "PULL-based (no streams)", "Bulk imports" (p. 988).
Sqoop streams every new row of a table into HDFS as soon as the row is inserted. False: there are "no streams" (p. 988); an incremental job fetches the new rows only when it is executed.

Show model answer

True: the first, the fourth and the fifth. The sink polls the channel; a memory channel is not persistent; Sqoop imports in bulk when it is run, not as a stream.

Each false statement swaps one word of a slide: push for poll, survives for "Not persistent", streams for "no streams". With a possible penalty, tick only what a slide sentence supports.

An exercise in writing a configuration and a command. Consider the ER model of the hospital heating management system of the Exam of 30 Jan 2026: a Hospital (Hospital_ID, Name) has Buildings (Building_ID, Name, Year), Departments (Department_ID, Specialty, N_Beds, Head_Physician_ID) and Heating Systems (Heater_Sys_ID, Heating_Power), which heat Floors. Every heating system posts a reading each minute as a JSON event over HTTP, with a header building that holds its Building_ID. The Department table, whose primary key is the integer Department_ID, lives in the hospital's MySQL database hospitaldb.

(a) Write the properties file of a Flume agent heatagent that receives the readings on port 8989 and writes them as plain text files into HDFS under /hospital/readings, one folder per building, without losing the readings held by the agent if its process restarts. (b) Write the Sqoop command that copies the Department table into the HDFS directory /hospital/departments with two mappers.

Show model answer

heatagent.sources = http-source
heatagent.sinks = hdfs-sink
heatagent.channels = ch

heatagent.sources.http-source.channels = ch
heatagent.sinks.hdfs-sink.channel = ch

heatagent.sources.http-source.type = org.apache.flume.source.http.HTTPSource
heatagent.sources.http-source.bind = 0.0.0.0
heatagent.sources.http-source.port = 8989

heatagent.sinks.hdfs-sink.type = hdfs
heatagent.sinks.hdfs-sink.hdfs.path = /hospital/readings/%{building}
heatagent.sinks.hdfs-sink.hdfs.filePrefix = readings
heatagent.sinks.hdfs-sink.hdfs.fileSuffix = .log
heatagent.sinks.hdfs-sink.hdfs.inUsePrefix = _
heatagent.sinks.hdfs-sink.hdfs.fileType = DataStream

heatagent.channels.ch.type = file

sqoop import \
--connect jdbc:mysql://localhost:3306/hospitaldb \
--username=admin \
--password=password \
--table=Department \
--target-dir /hospital/departments \
-m 2

(a) The agent is started with flume-ng agent --conf-file heatagent.properties --name heatagent. %{building} puts each reading in the folder of its header value; DataStream writes plain text; the file channel keeps the waiting readings on disk across a restart, which a memory channel would not. (b) The two mappers split the rows by ranges of the primary key Department_ID, so no --split-by is needed; each writes one file into the new directory.

The exercise is ours, in the setting of a past paper; no paper so far asks for Flume or Sqoop code. It tests three ideas: the wiring keys (channels on the source, channel on the sink), header bucketing in the HDFS path, and the file channel as the answer to "without losing". bind = 0.0.0.0 accepts posts from other machines; the slide's localhost accepts only local ones.

Check yourself

A count bolt with 3 tasks receives words by shuffle grouping. After a while, what does task 0 hold for the word "fresh"?

The exact number of times "fresh" appeared in the stream Only fields (or global) grouping would bring every "fresh" to one task; here tasks 1 and 2 got some of them.
Nothing, since shuffle grouping sends everything to the task with the lowest id That is global grouping.
A part of the count; the rest is spread over tasks 1 and 2 Shuffle picks a random task for each tuple, so every task holds a partial count — the widget showed 6 + 1 + 1.
Three times the true count, one copy per task Copies to every task are all grouping, and even then each task holds the true count once, not three times.

You raise a word count from 1 to 4 tasks per bolt and keep its groupings: shuffle into the split bolt, fields on "word" into the count bolt. Which statements hold? Select all that apply.

The split bolt still works: splitting a sentence needs no memory of other sentences A stateless bolt can take any tuple on any task — that is why shuffle grouping is fine there.
The counts are now spread over 4 tasks, so a final sum over the tasks is needed Not with fields grouping: each word is counted in one task only.
Global grouping on the count bolt would have done just as well The counts would be exact, but one task would do all the work and the other three would idle.
Each word is still counted by exactly one count task Fields grouping hashes the word, so the same word always reaches the same task, whatever the number of tasks.

Grouping follows state: stateless steps take shuffle, per-key state takes fields.

In a converging Flume flow of three tiers, HDFS goes down and stays down. In which order do the channels fill?

Tier 1 first, since it is the closest to the web servers Tier 1 can keep sending as long as tier 2 accepts; it fills last.
All tiers at the same time, since every tier carries the same volume Same throughput in steady state (p. 959) is not the same fill: only the tier that cannot deliver fills.
None of them: the sinks keep their events until HDFS is back Sinks hold nothing: a failed sink transaction rolls back, and the events stay in the channel.
Tier 3 first, then tier 2, then tier 1: back-pressure moves upstream one hop at a time The tier next to HDFS stops draining and fills; then its source refuses tier 2's batches, and so on — the widget's converging flow.

An agent with a memory channel holds 800 events when its process is killed. After the restart…

the 800 events are gone The memory channel is "Not persistent" (p. 968); the source had committed them, so the producers think they were delivered.
the sink delivers them, since its transaction had not committed The rollback protects against a failed delivery while the agent lives; the memory itself is gone.
the source asks the web servers to send them again The source had already committed them into the channel; nothing remembers them.
nothing is lost, because every channel writes to disk Only the file channel — the slides' "Disk" — does.

An agent has the two channels of p. 980 and the multiplexing selector of p. 982. An event arrives with the header show = 2. Where does it end up?

In ch1 and ch2: every event goes everywhere That is the replicating selector, the default when no selector is configured.
Only in ch2, so it is printed by the logger sink and never reaches HDFS "If show = 2, route ch2" (p. 982), and ch2 is drained by log-sink (p. 980).
In ch1, because the selector reads the body of the event The selector reads a header, here show; the body is never looked at.
Wherever the HDFS sink decides The selector belongs to the source; each sink just drains its one channel.

You run sqoop import with -m 4 on a table that has no primary key, and no --split-by. What happens?

Sqoop splits on the first column of the table Sqoop does not guess a split column.
Four mappers each read the whole table That would copy every row four times; Sqoop refuses instead.
It works, because the primary key only decides the number of mappers The other way round: -m sets the number, the key the split — the red pen of p. 991.
The import fails: give a split column with --split-by, or run one mapper with -m 1 Without a column to cut into ranges there is no way to share the rows among four mappers.

Which of these situations call for Sqoop rather than Flume? Select all that apply.

The data is a customers table in a MySQL database Relational tables are Sqoop's input, through a JDBC connector.
Web servers write log lines all day long A never-ending stream of events: Flume.
A copy once a night is enough, and only the rows changed since yesterday are needed A saved job with --incremental lastmodified and a check column.
Events must land in different HDFS folders according to one of their headers Bucketing with %{header} in the path of Flume's HDFS sink.

Key take-aways

Batch tools read inputs that have ended; a stream is an unbounded sequence of tuples, queried continuously over windows. Storm computes on streams with topologies — graphs of spouts (sources of streams) and bolts (processing) that run until killed — on a cluster where Nimbus assigns, the Supervisors run and ZooKeeper coordinates.
Every component runs as many tasks; the stream grouping of each edge decides which task gets a tuple: shuffle (random), fields (same values, same task — needed for per-key state such as a word count), all (a copy to every task), global (the task with the lowest id). Storm's "no data loss" is at-least-once.
A Flume agent is source → channel → sink: the source accepts data and pushes events (headers + body) into the channel, which stores them in the order received; the sink polls, removes and delivers. A source may feed many channels; a sink reads one.
Transactions on both sides of the channel, with the sink committing only after delivery, prevent loss in transit — with a durable (file) channel, at-least-once, and with weak ordering. A memory channel is fast but loses its events in a crash.
Converging flows are sized with three rules: 4 to 16 client agents per aggregator; exit batch size = tier exit volume / agents in the tier; channel capacity = worst ingest rate × worst outage. A properties file declares names, wires sources to channels and sinks to one channel each, and sets types: HDFS sink with path, prefixes and file type, bucketing by header or time, replicating or multiplexing selectors, interceptors.
Sqoop pulls tables or query results in bulk through JDBC, as a map-only MapReduce job: -m mappers on ranges of the primary key or --split-by; $CONDITIONS in a --query; --hive-import into a managed Hive table; saved incremental jobs with a check column and a last value.

Stream (p. 927): unbounded sequence of tuples; a tuple is a named list of values
Spout · bolt (pp. 928–929): source of streams · processes input streams and produces new streams (filters, aggregation, join)
Topology (p. 930): network of spouts and bolts; runs until killed; components execute as many tasks (p. 931)
Storm cluster (p. 925): Nimbus: master node, similar to the JobTracker · ZooKeeper: cluster coordination, holds the state · Supervisors: run worker processes
Stream grouping (p. 933): shuffle: random task · fields: hash of fields, same values → same task · all: every task · global: lowest task id
Storm guarantee: "no data loss" = at-least-once: tuples replayed, duplicates possible
Flume: "a continuous data ingestion system" (p. 939), born for log aggregation at Cloudera
Agent (p. 948): source (accepts, writes to channel) → channel (stores in order received) → sink (removes, sends downstream)
Event (p. 947): header: metadata, used for routing · body: actual content
Push · poll (p. 946): source pushes events · sink polls for events
Cardinalities (p. 966): source–channel many-to-many · channel–sink one-to-many: a sink reads one channel
Channels (p. 968): memory: in-memory queue, fast, not persistent, limited capacity · Disk = file channel (type = file): durable
Ordering (p. 986): weak: in the order received, in the absence of failures
Tiers (p. 961): one aggregating agent per 4 to 16 client agents
Exit batch size (p. 962): $B_t = V / N_t$: tier exit volume / agents in the tier
Channel capacity (p. 963): $C = r_{\text{in}}^{\max} \times T_{\text{out}}^{\max}$: worst ingest rate × worst outage
Start an agent (p. 970): flume-ng agent --conf-file spool-to-logger.properties --name agent1
Wiring (p. 971): agent1.sources.source1.channels = channel1 (plural) · agent1.sinks.sink1.channel = channel1 (singular)
HDFS sink (pp. 972–974): type = hdfs, hdfs.path, hdfs.filePrefix, hdfs.fileSuffix, hdfs.inUsePrefix = _, hdfs.fileType = DataStream; rolls "every 30 seconds" (p. 973) or earlier, at the first of rollInterval, rollSize, rollCount to fire
Bucketing (pp. 977–978): hdfs.path = /tmp/flume/http/%{topic} · …/%Y/%m/%d/%H with hdfs.useLocalTimeStamp = true
Selectors (pp. 981–982): replicating: default, every event to every channel · selector.type = multiplexing, selector.header = show, selector.mapping.1 = ch1
Interceptor (pp. 984–985): sources.source1.interceptors = regexInterceptor, then ….regexInterceptor.type = regex_filter, regex = .*election.*, excludeEvents = false
Sqoop (p. 988): import from relational DBs; pull-based, no streams; bulk; one command
Import (p. 990): sqoop import --connect jdbc:mysql://localhost:3306/nseProd --username=qt --password=password --table=tradingDays --target-dir /mysql/nseProd --m 1
Mappers and split: -m mappers (default 4); ranges of the primary key or of --split-by col; -m 1 when there is no key and no split column
Query import (p. 993): --query '… where … and $CONDITIONS' with --target-dir, and --split-by or -m 1
Hive (p. 994): --hive-import --hive-table=tradingDays: a managed table, files moved into the warehouse
Incremental job (pp. 995–999): sqoop job --create myjob -- import … --append --incremental lastmodified --check-column ts · sqoop job --exec myjob; next run gets data > LastValue; --append lets each run write into the existing target directory

What to watch and read

History of Apache Storm and lessons learned Nathan Marz, creator of Storm
How Storm was built at BackType, bought with it by Twitter, and released in 2011 — the story behind the first line of p. 924, told by Storm's creator, whom the deck thanks on p. 1000.
Storm Tutorial Apache Storm documentation
Nimbus, Supervisors and ZooKeeper, streams, spouts and bolts, and the word-count topology with its shuffle and fields groupings: pp. 925–933 in prose, with the code of section 2.
Flume User Guide Apache Flume project
The source of the Flume slides: the agent and multi-agent figures, the catalogues of sources, channels and sinks with every property, selectors and interceptors. Keep it open next to section 5 and the widget's generated file.
Sqoop User Guide (v1.4.7) Apache Sqoop project
Controlling parallelism, free-form query imports with $CONDITIONS, Hive imports, incremental imports and saved jobs — the sentences the red pens of section 6 quote. Sqoop has since retired to the Apache Attic; the ideas live on in Spark's JDBC reader.
Streaming 101: The world beyond batch Tyler Akidau, O'Reilly Radar
Bounded against unbounded data, event time against processing time, and fixed, sliding and session windows, with figures: the frame of the course notes' chapter on streaming, taken further.

Streams into Hadoop: Storm, Flume and Sqoop

A computation that never finishes

Four words: stream, spout, bolt, topology

The cluster

Which task gets the tuple?

Topology playground: where does each word go?

Flume: a pond at every station

The agent: source, channel, sink

Transactions

Why channels?

From three web servers to a converging flow

Three rules of thumb

An agent in ten lines of properties

The HDFS sink

HTTP in, folders by header

Several channels, selectors and interceptors

Flume agent builder: wire a flow, then hit it

Sqoop: the nightly truck

How the rows are split

Into Hive, and only what is new

Exam-style questions

Check yourself

Key take-aways

What to watch and read