Kafka-HDFS Ingestion Error

Lee kyo

unread,

Feb 16, 2016, 10:30:56 AM2/16/16

to gobblin-users

hello.

For my experiment, I have

--- 3 kafka brokers

--- 5 nodes for hadoop cluster

--- a Kafka topic with txt format

--- gobblin 0.6.2

My job configuration file looks like

job.name=GobblinKafkaQuickStart2

job.group=GobblinKafka

job.description=Gobblin quick start job for Kafka

job.lock.enabled=false

kafka.brokers=[broker1 ip]:9092,[broker2 ip]:9092,[broker3 ip]:9092

topic.whitelist=test

bootstrap.with.offset=earliest

topic.name=test

source.class=gobblin.source.extractor.extract.kafka.KafkaSimpleSource

extract.namespace=gobblin.extract.kafka

writer.builder.class=gobblin.writer.SimpleDataWriterBuilder

writer.file.path.type=tablename

writer.destination.type=HDFS

writer.output.format=txt

data.publisher.type=gobblin.publisher.BaseDataPublisher

mr.job.max.mappers=1

metrics.reporting.file.enabled=true

metrics.log.dir=/gobblin-kafka/metrics

metrics.reporting.file.suffix=txt

bootstrap.with.offset=earliest

fs.uri=hdfs://dev-likehadoop001.ncl:9000

writer.fs.uri=hdfs://dev-likehadoop001.ncl:9000

#state.store.fs.uri=hdfs://dev-likehadoop001.ncl:9000

mr.job.root.dir=/user/irteam/gobblin-kafka/working

state.store.dir=/user/irteam/gobblin-kafka/state-store

task.data.root.dir=/user/irteam/jobs/kafkaetl/gobblin/gobblin-kafka/task-data

data.publisher.final.dir=/user/irteam/gobblintest/job-output

The following command is used to submit the job:

bin/gobblin-mapreduce.sh --conf job/kafka2.pull

Server log is look like :

SLF4J: Class path contains multiple SLF4J bindings.

SLF4J: Found binding in [jar:file:/home1/irteam/apps/gobblin/gobblin-dist/lib/slf4j-log4j12-1.7.5.jar!/org/slf4j/impl/StaticLoggerBinder.class]

SLF4J: Found binding in [jar:file:/home1/irteam/apps/hadoop-2.6.0/share/hadoop/common/lib/slf4j-log4j12-1.7.5.jar!/org/slf4j/impl/StaticLoggerBinder.class]

SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an explanation.

SLF4J: Actual binding is of type [org.slf4j.impl.Log4jLoggerFactory]

WARN [JobContext] Property writer.staging.dir is deprecated. No need to use it if task.data.root.dir is specified.

WARN [JobContext] Property writer.output.dir is deprecated. No need to use it if task.data.root.dir is specified.

WARN [MRJobLauncher] Job working directory already exists for job GobblinKafkaQuickStart2

WARN [KafkaSource] Previous offset for partition test:0 does not exist. This partition will be skipped.

WARN [KafkaSource] Previous offset for partition test:1 does not exist. This partition will be skipped.

WARN [KafkaSource] Previous offset for partition test:2 does not exist. This partition will be skipped.

WARN [KafkaSource] Previous offset for partition test:3 does not exist. This partition will be skipped.

WARN [KafkaSource] Previous offset for partition test:4 does not exist. This partition will be skipped.

WARN [KafkaSource] Previous offset for partition test:5 does not exist. This partition will be skipped.

WARN [KafkaSource] Previous offset for partition test:6 does not exist. This partition will be skipped.

WARN [KafkaSource] Previous offset for partition test:7 does not exist. This partition will be skipped.

WARN [KafkaSource] Previous offset for partition test:8 does not exist. This partition will be skipped.

WARN [KafkaSource] Previous offset for partition test:9 does not exist. This partition will be skipped.

Error: java.lang.ClassNotFoundException: com.typesafe.config.ConfigFactory

at java.net.URLClassLoader$1.run(URLClassLoader.java:366)

at java.net.URLClassLoader$1.run(URLClassLoader.java:355)

at java.security.AccessController.doPrivileged(Native Method)

at java.net.URLClassLoader.findClass(URLClassLoader.java:354)

at java.lang.ClassLoader.loadClass(ClassLoader.java:424)

at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:308)

at java.lang.ClassLoader.loadClass(ClassLoader.java:357)

at gobblin.util.ConfigUtils.propertiesToConfig(ConfigUtils.java:95)

at gobblin.metrics.reporter.OutputStreamReporter$Builder.build(OutputStreamReporter.java:162)

at gobblin.metrics.GobblinMetrics.buildFileMetricReporter(GobblinMetrics.java:468)

at gobblin.metrics.GobblinMetrics.startMetricReporting(GobblinMetrics.java:380)

at gobblin.metrics.GobblinMetrics.startMetricReportingWithFileSuffix(GobblinMetrics.java:340)

at gobblin.runtime.mapreduce.MRJobLauncher$TaskRunner.setup(MRJobLauncher.java:533)

at gobblin.runtime.mapreduce.MRJobLauncher$TaskRunner.run(MRJobLauncher.java:540)

at org.apache.hadoop.mapred.MapTask.runNewMapper(MapTask.java:784)

at org.apache.hadoop.mapred.MapTask.run(MapTask.java:341)

at org.apache.hadoop.mapred.YarnChild$2.run(YarnChild.java:163)

at java.security.AccessController.doPrivileged(Native Method)

at javax.security.auth.Subject.doAs(Subject.java:415)

at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1628)

at org.apache.hadoop.mapred.YarnChild.main(YarnChild.java:158)