Tom White - Hadoop The Definitive Guide_ 4 edition - 2015 (811394), страница 81
Текст из файла (страница 81)
The Avro sink-source pair is older than the Thrift equivalent, and (at the time of writing) has some featuresthat the Thrift one doesn’t provide, such as encryption.Distribution: Agent Tiers|391Example 14-4 shows a two-tier Flume configuration. Two agents are defined in the file,named agent1 and agent2. An agent of type agent1 runs in the first tier, and has aspooldir source and an Avro sink connected by a file channel.
The agent2 agent runsin the second tier, and has an Avro source that listens on the port that agent1’s Avrosink sends events to. The sink for agent2 uses the same HDFS sink configuration fromExample 14-2.Notice that since there are two file channels running on the same machine, they areconfigured to point to different data and checkpoint directories (they are in the user’shome directory by default). This way, they don’t try to write their files on top of oneanother.Example 14-4.
A two-tier Flume configuration using a spooling directory source and anHDFS sink# First-tier agentagent1.sources = source1agent1.sinks = sink1agent1.channels = channel1agent1.sources.source1.channels = channel1agent1.sinks.sink1.channel = channel1agent1.sources.source1.type = spooldiragent1.sources.source1.spoolDir = /tmp/spooldiragent1.sinks.sink1.type = avroagent1.sinks.sink1.hostname = localhostagent1.sinks.sink1.port = 10000agent1.channels.channel1.type = fileagent1.channels.channel1.checkpointDir=/tmp/agent1/file-channel/checkpointagent1.channels.channel1.dataDirs=/tmp/agent1/file-channel/data# Second-tier agentagent2.sources = source2agent2.sinks = sink2agent2.channels = channel2agent2.sources.source2.channels = channel2agent2.sinks.sink2.channel = channel2agent2.sources.source2.type = avroagent2.sources.source2.bind = localhostagent2.sources.source2.port = 10000agent2.sinks.sink2.type = hdfsagent2.sinks.sink2.hdfs.path = /tmp/flumeagent2.sinks.sink2.hdfs.filePrefix = events392|Chapter 14: Flumeagent2.sinks.sink2.hdfs.fileSuffix = .logagent2.sinks.sink2.hdfs.fileType = DataStreamagent2.channels.channel2.type = fileagent2.channels.channel2.checkpointDir=/tmp/agent2/file-channel/checkpointagent2.channels.channel2.dataDirs=/tmp/agent2/file-channel/dataThe system is illustrated in Figure 14-4.Figure 14-4.
Two Flume agents connected by an Avro sink-source pairEach agent is run independently, using the same --conf-file parameter but differentagent --name parameters:% flume-ng agent --conf-file spool-to-hdfs-tiered.properties --name agent1 ...and:% flume-ng agent --conf-file spool-to-hdfs-tiered.properties --name agent2 ...Delivery GuaranteesFlume uses transactions to ensure that each batch of events is reliably delivered from asource to a channel, and from a channel to a sink. In the context of the Avro sink-sourceconnection, transactions ensure that events are reliably delivered from one agent to thenext.The operation to read a batch of events from the file channel in agent1 by the Avro sinkwill be wrapped in a transaction. The transaction will only be committed once the AvroDistribution: Agent Tiers|393sink has received the (synchronous) confirmation that the write to the Avro source’sRPC endpoint was successful.
This confirmation will only be sent once agent2’s trans‐action wrapping the operation to write the batch of events to its file channel has beensuccessfully committed. Thus, the Avro sink-source pair guarantees that an event isdelivered from one Flume agent’s channel to another Flume agent’s channel (at leastonce).If either agent is not running, then clearly events cannot be delivered to HDFS. Forexample, if agent1 stops running, then files will accumulate in the spooling directory,to be processed once agent1 starts up again. Also, any events in an agent’s own filechannel at the point the agent stopped running will be available on restart, due to thedurability guarantee that file channel provides.If agent2 stops running, then events will be stored in agent1’s file channel until agent2starts again. Note, however, that channels necessarily have a limited capacity; ifagent1’s channel fills up while agent2 is not running, then any new events will be lost.By default, a file channel will not recover more than one million events (this can beoverridden by its capacity property), and it will stop accepting events if the free diskspace for its checkpoint directory falls below 500 MB (controlled by the minimumRequiredSpace property).Both these scenarios assume that the agent will eventually recover, but that is not alwaysthe case (if the hardware it is running on fails, for example).
If agent1 doesn’t recover,then the loss is limited to the events in its file channel that had not been delivered toagent2 before agent1 shut down. In the architecture described here, there are multiplefirst-tier agents like agent1, so other nodes in the tier can take over the function of thefailed node. For example, if the nodes are running load-balanced web servers, then othernodes will absorb the failed web server’s traffic, and they will generate new Flume eventsthat are delivered to agent2. Thus, no new events are lost.An unrecoverable agent2 failure is more serious, however.
Any events in the channelsof upstream first-tier agents (agent1 instances) will be lost, and all new events generatedby these agents will not be delivered either. The solution to this problem is for agent1to have multiple redundant Avro sinks, arranged in a sink group, so that if the destinationagent2 Avro endpoint is unavailable, it can try another sink from the group. We’ll seehow to do this in the next section.394|Chapter 14: FlumeSink GroupsA sink group allows multiple sinks to be treated as one, for failover or load-balancingpurposes (see Figure 14-5).
If a second-tier agent is unavailable, then events will bedelivered to another second-tier agent and on to HDFS without disruption.Figure 14-5. Using multiple sinks for load balancing or failoverTo configure a sink group, the agent’s sinkgroups property is set to define the sinkgroup’s name; then the sink group lists the sinks in the group, and also the type of thesink processor, which sets the policy for choosing a sink.
Example 14-5 shows the con‐figuration for load balancing between two Avro endpoints.Example 14-5. A Flume configuration for load balancing between two Avro endpointsusing a sink groupagent1.sources = source1agent1.sinks = sink1a sink1bagent1.sinkgroups = sinkgroup1agent1.channels = channel1agent1.sources.source1.channels = channel1agent1.sinks.sink1a.channel = channel1agent1.sinks.sink1b.channel = channel1agent1.sinkgroups.sinkgroup1.sinks = sink1a sink1bagent1.sinkgroups.sinkgroup1.processor.type = load_balanceagent1.sinkgroups.sinkgroup1.processor.backoff = trueSink Groups|395agent1.sources.source1.type = spooldiragent1.sources.source1.spoolDir = /tmp/spooldiragent1.sinks.sink1a.type = avroagent1.sinks.sink1a.hostname = localhostagent1.sinks.sink1a.port = 10000agent1.sinks.sink1b.type = avroagent1.sinks.sink1b.hostname = localhostagent1.sinks.sink1b.port = 10001agent1.channels.channel1.type = fileThere are two Avro sinks defined, sink1a and sink1b, which differ only in the Avroendpoint they are connected to (since we are running all the examples on localhost, itis the port that is different; for a distributed install, the hosts would differ and the portswould be the same).
We also define sinkgroup1, and set its sinks to sink1a and sink1b.The processor type is set to load_balance, which attempts to distribute the event flowover both sinks in the group, using a round-robin selection mechanism (you can changethis using the processor.selector property). If a sink is unavailable, then the next sinkis tried; if they are all unavailable, the event is not removed from the channel, just likein the single sink case. By default, sink unavailability is not remembered by the sinkprocessor, so failing sinks are retried for every batch of events being delivered. This canbe inefficient, so we have set the processor.backoff property to change the behaviorso that failing sinks are blacklisted for an exponentially increasing timeout period (upto a maximum period of 30 seconds, controlled by processor.selector.maxTimeOut).There is another type of processor, failover, that instead of loadbalancing events across sinks uses a preferred sink if it is available,and fails over to another sink in the case that the preferred sink isdown.
The failover sink processor maintains a priority order for sinksin the group, and attempts delivery in order of priority. If the sinkwith the highest priority is unavailable the one with the next highestpriority is tried, and so on. Failed sinks are blacklisted for an increas‐ing timeout period (up to a maximum period of 30 seconds, con‐trolled by processor.maxpenalty).The configuration for one of the second-tier agents, agent2a, is shown in Example 14-6.396|Chapter 14: FlumeExample 14-6. Flume configuration for second-tier agent in a load balancing scenarioagent2a.sources = source2aagent2a.sinks = sink2aagent2a.channels = channel2aagent2a.sources.source2a.channels = channel2aagent2a.sinks.sink2a.channel = channel2aagent2a.sources.source2a.type = avroagent2a.sources.source2a.bind = localhostagent2a.sources.source2a.port = 10000agent2a.sinks.sink2a.type = hdfsagent2a.sinks.sink2a.hdfs.path = /tmp/flumeagent2a.sinks.sink2a.hdfs.filePrefix = events-aagent2a.sinks.sink2a.hdfs.fileSuffix = .logagent2a.sinks.sink2a.hdfs.fileType = DataStreamagent2a.channels.channel2a.type = fileThe configuration for agent2b is the same, except for the Avro source port (since weare running the examples on localhost) and the file prefix for the files created by theHDFS sink.
![](https://s.studizba.com/z.php?f=/templates/si/i/noavatar.png&w=100&h=100&t=1"){kind=avatar}
