Я хочу записать данные в каталог хранилища улья, в две отдельные таблицы с именами flumemaleemployee и flumefemaleemployee. Последние 3 records должны быть вставлены в таблицу female, а верхние 3 records должны быть вставлены в таблицу male. Ниже приведены мои данные:
1,alok,mumbai
1,jatin,chennai
1,yogesh,kolkata
2,ragini,delhi
2,jyotsana,pune
1,valmiki,banglore
Ниже мой код flume conf:
agent.sources = tailsrc
agent.channels = mem1 mem2
agent.sinks = stdl std2
agent.sources.tailsrc.type = exec
agent.sources.tailsrc.command = tail -F /home/cloudera/Desktop/in.txt
agent.sources.tailsrc.batchSize = 1
agent.sources.tailsrc.interceptors = i1
agent.sources.tailsrc.interceptors.i1.type = regex_extractor
agent.sources.tailsrc.interceptors.il.regex = A(\\d}
agent.sources.tailsrc. interceptors. M.serializers = t1
agent.sources.tailsrc. interceptors, i1.serializers.t1. name = type
agent.sources.tailsrc.selector.type = multiplexing
agent.sources.tailsrc.selector.header = type
agent.sources.tailsrc.selector.mapping.1 = mem1
agent.sources.tailsrc.selector.mapping.2 = mem2
agent.sinks.std1.type = hdfs
agent.sinks.stdl.channel = mem1
agent.sinks.stdl.batchSize = 1
agent.sinks.std1.hdfs.path = /user/hive/warehouse/aisehibanayatp.db/flumemaleemployee
agent.sinks.stdl.rolllnterval = 0
agent.sinks.stdl.hdfs.fileType = DataStream
agent.sinks.std2.type = hdfs
agent.sinks.std2.channel = mem2
agent.sinks.std2.batchSize = 1
agent.sinks.std2.hdfs.path = /user/hi ve/warehouse/aisehibanayatp.db/flumefemaleemployee
agent.sinks.std2.rolllnterval = 0
agent.sinks.std2.hdfs.fileType = DataStream
agent.channels.mem1.type = memory
agent.channels.meml.capacity = 100
agent.channels.mem2.type = memory
agent.channels.mem2.capacity = 100
agent.sources.tailsrc.channels = mem1 mem2
Я не получаю никаких ошибок, но когда я запускаю flume service с помощью приведенной ниже команды, он просто застревает на чем-то, что я не знаю, как с этим справиться, так как я не получаю никаких ошибок
flume-ng agent --name agent -conf-file /home/cloudera/Desktop/flume1.config
и он застрял на шаге ниже:
18/11/13 08:03:00 INFO instrumentation.MonitoredCounterGroup: Shutdown Metric for type: CHANNEL, name: mem2. channel.event.take.success == 0
18/11/13 08:03:00 INFO node.Application: Starting new configuration:{ sourceRunners:{} sinkRunners:{std2=SinkRunner: { policy:org.apache.flume.sink.DefaultSinkProcessor@17ade71c counterGroup:{ name:null counters:{} } }} channels:{mem2=org.apache.flume.channel.MemoryChannel{name: mem2}} }
18/11/13 08:03:00 INFO node.Application: Starting Channel mem2
18/11/13 08:03:00 INFO instrumentation.MonitoredCounterGroup: Component type: CHANNEL, name: mem2 started
18/11/13 08:03:00 INFO node.Application: Starting Sink std2
18/11/13 08:03:00 INFO instrumentation.MonitoredCounterGroup: Monitored counter group for type: SINK, name: std2: Successfully registered new MBean.
18/11/13 08:03:00 INFO instrumentation.MonitoredCounterGroup: Component type: SINK, name: std2 started
Итак, как я могу этого добиться??
