Data ETL tools for hadoop ecosystem Morphlines

撒的阿坎努斯

　　when i use
　　there is a error

java.lang.NoClassDefFoundError: org/kitesdk/morphline/api/MorphlineCompilationException
at java.lang.Class.forName0(Native Method)
at java.lang.Class.forName(Class.java:190)
at org.apache.flume.sink.solr.morphline.MorphlineSink.start(MorphlineSink.java:93)
at org.apache.flume.sink.DefaultSinkProcessor.start(DefaultSinkProcessor.java:46)
at org.apache.flume.SinkRunner.start(SinkRunner.java:79)
at org.apache.flume.lifecycle.LifecycleSupervisor$MonitorRunnable.run(LifecycleSupervisor.java:251)
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:471)
at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:304)

　　solution:

* clone flume from git
* cd flume
* edit flume-ng-sinks/flume-ng-morphline-solr-sink/pom.xml, in there do the following:
** make cdk-morphlines-all required by commenting out this blurb: <optional>true</optional>
** add the following mvn blurb to the <build> element in order to copy the dependency jars into the target/lib dir:
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-dependency-plugin</artifactId>
<executions>
<execution>
<phase>package</phase>
<goals>
<goal>copy-dependencies</goal>
</goals>
<configuration>
<outputDirectory>${project.build.directory}/lib</outputDirectory>
<includeScope>runtime</includeScope> <!-- excludes test jars; see http://jira.codehaus.org/browse/MDEP-128 -->
<excludeScope>provided</excludeScope>
</configuration>
</execution>
</executions>
</plugin>
* mvn -Dhadoop.profile=2 clean package -pl flume-ng-sinks/flume-ng-morphline-solr-sink
* find flume-ng-sinks/flume-ng-morphline-solr-sink/target -name '*.jar'
* copy the jars printed out by the above find command into the flume lib dir

　　see : https://groups.google.com/a/cloudera.org/forum/#!msg/cdk-dev/7T4pTebdWN4/sBHGkoS70LkJ
　　---------
　　solr server's version is 4.10.1
　　error
　　

 
　　solution:
　　find solr's lib

ls /home/tomcat/solr/WEB-INF/lib/
antlr-runtime-3.5.jar                hadoop-hdfs-2.2.0.jar                  lexicon                               lucene-suggest-4.10.1.jar
asm-4.1.jar                          hppc-0.5.2.jar                         log4j-1.2.16.jar                      mahout-collections-1.0.jar
asm-commons-4.1.jar                  httpclient-4.3.1.jar                   lucene-analyzers-common-4.10.1.jar    mahout-math-0.6.jar
attributes-binder-1.2.2.jar          httpcore-4.3.jar                       lucene-analyzers-kuromoji-4.10.1.jar  noggit-0.5.jar
carrot2-core-3.10.0-SNAPSHOT.jar     httpmime-4.3.1.jar                     lucene-analyzers-phonetic-4.10.1.jar  org.restlet-2.1.1.jar
commons-cli-1.2.jar                  inok-solr-dataimportscheduler-1.1.jar  lucene-analyzers-smartcn-4.10.1.jar   org.restlet.ext.servlet-2.1.1.jar
commons-codec-1.9.jar                jackson-core-asl-1.9.13.jar            lucene-codecs-4.10.1.jar              protobuf-java-2.5.0.jar
commons-configuration-1.6.jar        jackson-mapper-asl-1.9.13.jar          lucene-core-4.10.1.jar                simple-xml-2.7.jar
commons-fileupload-1.2.1.jar         jcl-over-slf4j-1.6.6.jar               lucene-expressions-4.10.1.jar         slf4j-api-1.6.6.jar
commons-io-2.3.jar                   jcseg-analyzer-1.9.5.jar               lucene-grouping-4.10.1.jar            slf4j-log4j12-1.6.6.jar
commons-lang-2.6.jar                 jcseg-core-1.9.5.jar                   lucene-highlighter-4.10.1.jar         solr-core-4.10.1.jar
concurrentlinkedhashmap-lru-1.2.jar  jcseg-core-1.9.5.jar.old               lucene-join-4.10.1.jar                solr-solrj-4.10.1.jar
dom4j-1.6.1.jar                      jcseg.properties                       lucene-memory-4.10.1.jar              spatial4j-0.4.1.jar
guava-14.0.1.jar                     jcseg-solr-1.9.5.jar                   lucene-misc-4.10.1.jar                wstx-asl-3.2.7.jar
hadoop-annotations-2.2.0.jar         jcseg-solr-1.9.5.jar.old               lucene-queries-4.10.1.jar             zookeeper-3.4.6.jar
hadoop-auth-2.2.0.jar                joda-time-2.2.jar                      lucene-queryparser-4.10.1.jar
hadoop-common-2.2.0.jar              jul-to-slf4j-1.6.6.jar                 lucene-spatial-4.10.1.jar

　　#cp  /home/tomcat/solr/WEB-INF/lib/lucene-*-4.10.1.jar   /root/aflume/lib/
　　#cp /home/tomcat/solr/WEB-INF/lib/solr-*-4.10.1.jar    /root/aflume/lib/
　　#cp /home/tomcat/solr/WEB-INF/lib/http*-4.3.1.jar   /root/aflume/lib/
　　delete old related jars in /root/aflume/lib/  such as lucene-*-4.3.0.jar  solr-*-4.3.0.jar   http*-4.2.1.jar
　　-------
　　when run
　　#flume-ng agent -c ./conf/ -f conf/flume.conf -Dflume.root.logger=DEBUG,console -n a1
　　flume/conf/flume.conf

# Define a memory channel called ch1 on agent1
a1.channels.ch1.type = memory
# Define an Avro source called avro-source1 on agent1 and tell it
# to bind to 0.0.0.0:41414. Connect it to channel ch1.
a1.sources.mysqlbinlog.channels = ch1
a1.sources.mysqlbinlog.type = com.inoknok.mysqlbinlog.source.InokMysqlBinlogSource
a1.sources.mysqlbinlog.user=zhaohj
a1.sources.mysqlbinlog.password=zhaohj111
a1.sources.mysqlbinlog.host=192.168.0.135
a1.sources.mysqlbinlog.port=3306
a1.sources.mysqlbinlog.serverId=2
a1.sources.mysqlbinlog.autoReconnect=false

# Define a logger sink that simply logs all events it receives
# and connect it to the other end of the same channel.
a1.sinks.k1.channel = ch1
a1.sinks.k1.type = org.apache.flume.sink.solr.morphline.MorphlineSolrSink
a1.sinks.k1.morphlineFile = /root/aflume/conf/morphline.conf
# Finally, now that we've defined all of our components, tell
# agent1 which ones we want to activate.
a1.channels = ch1
a1.sources = mysqlbinlog
a1.sinks = k1

　　flume/conf/morphile.conf

morphlines : [
{
id : morphline1
importCommands : ["org.kitesdk.**", "com.cloudera.**", "org.apache.solr.**"]
commands : [
{
readClob {
charset : UTF-8
}
}
{
split {
inputField : message
outputFields : ["", "", "",user_id, username, age]
separator : "\u0001"
isRegex : false
addEmptyStrings : false
trim : true
}
}
{
sanitizeUnknownSolrFields {
solrLocator : {
solrUrl : "192.168.10.204:8983"
solrHomeDir : "/home/tomcat/solrHome/test.inok_user"
#collection : "@${tablename}"
#zkHost: "192.168.10.204:8983"
}
}
}
{
logDebug {
format : "xxxxxxxxx   xxxxxx My output record: {}"
args : ["@{}"]
}
}

# load the record into a Solr server or MapReduce Reducer
{
loadSolr {
solrLocator : {
solrUrl : "192.168.10.204:8983"
solrHomeDir : "/home/tomcat/solrHome/test.inok_user"
#collection : "@${tablename}"       # Name of solr collection
#zkHost : "192.168.10.204:8983" # ZooKeeper ensemble
}
}
}
]
}
]

　　error:
　　

 
　　Solution:

        solrUrl : "http://192.168.10.204:8983/test.inok_user"
solrHomeDir : "/home/tomcat/solrHome/test.inok_user"
　　----------
　　Morphline - Choose collection for loadSolr at run time
　　https://groups.google.com/a/cloudera.org/forum/#!topic/search-user/z9A_Xe5FviM
　　http://kitesdk.org/docs/current/morphlines/morphlines-reference-guide.html#Implementing_your_own_Custom_Command
　　Use case: I want to dynaic set solrLocator at run time, but faild. The replacement way is to use tryRule command, but the solrLocator must be setted hardcode.

morphlines : [
{
id : morphline1
importCommands : ["org.kitesdk.**", "com.cloudera.**", "org.apache.solr.**"]
commands : [
{
readClob {
charset : UTF-8
}
}

{
tryRules {
catchExceptions : false
throwExceptionIfAllRulesFailed : true
rules : [
{
//test.inok_user
commands : [
{contains {tablename : [test.inok_user]} }
{logDebug { format : "YYYYYYYYY 1111 My output record: {}, talbename={}",args : ["@{}","@{tablename}"]} }
{split {inputField : message, outputFields : [updatetime,"","",user_id, username, age],separator : "\u0001",
isRegex : false,addEmptyStrings : false, trim : true} }
{convertTimestamp {
field : updatetime
inputFormats : ["unixTimeInMillis"]
inputTimezone : UTC
outputFormat : "yyyy-MM-dd'T'HH:mm:ss.SSS'Z'"
outputTimezone : PRC
}
}
{sanitizeUnknownSolrFields {
solrLocator : {
solrUrl :"http://192.168.10.204:8983/test.inok_user/"
solrHomeDir : "/home/tomcat/solrHome/test.inok_user"
}
}}
{logDebug { format : "xxxxxxxxx 1111 My output record: {}, talbename={}",args : ["@{}","@{tablename}"]}  }
# load the record into a Solr server or MapReduce Reducer
{loadSolr { solrLocator : {
solrUrl : "http://192.168.10.204:8983/test.inok_user"
solrHomeDir : "/home/tomcat/solrHome/test.inok_user"
batchSize : 1
}  }  }
]
}//end rule1
#test.inoktest
{
//test.inoktest
commands : [
{contains {tablename : [test.inoktest]} }
{logDebug {format : "YYYYYYYYY 2222 My output record: {}, talbename={}",args : ["@{}","@{tablename}"] } }
{split {inputField : message, outputFields : [updatetime,"","",id, content],separator : "\u0001",
isRegex : false,addEmptyStrings : false, trim : true} }
{convertTimestamp {
field : updatetime
inputFormats : ["unixTimeInMillis"]
inputTimezone : UTC
outputFormat : "yyyy-MM-dd'T'HH:mm:ss.SSS'Z'"
outputTimezone : PRC
}
}
{sanitizeUnknownSolrFields {solrLocator : {
solrUrl :"http://192.168.10.204:8983/test.inoktest/"
solrHomeDir : "/home/tomcat/solrHome/test.inoktest"
}
}}
{logDebug { format : "xxxxxxxxx 2222 My output record: {}, talbename={}",args : ["@{}","@{tablename}"]} }
# load the record into a Solr server or MapReduce Reducer
{loadSolr { solrLocator : {
solrUrl : "http://192.168.10.204:8983/test.inoktest"
solrHomeDir : "/home/tomcat/solrHome/test.inoktest"
batchSize : 1
}
} }
]
}//end rule2
]
}//end rules
}
]
}
]

　　--------------------
　　when I configure avro source in agent a while an avro sink in agent b in different hosts.
　　flume.conf in host a

# Define a memory channel called ch1 on agent1
a1.channels.ch1.type = memory
# Define an Avro source called avro-source1 on agent1 and tell it
# to bind to 0.0.0.0:41414. Connect it to channel ch1.
a1.sources.mysqlbinlog.channels = ch1
a1.sources.mysqlbinlog.type = com.inoknok.mysqlbinlog.source.InokMysqlBinlogSource
a1.sources.mysqlbinlog.user=zhaohj
a1.sources.mysqlbinlog.password=zhaohj111
a1.sources.mysqlbinlog.host=192.168.0.135
a1.sources.mysqlbinlog.port=3306
a1.sources.mysqlbinlog.serverId=2
a1.sources.mysqlbinlog.autoReconnect=false

#sink to solr
a1.sinks.k1.channel = ch1
a1.sinks.k1.type = org.apache.flume.sink.solr.morphline.MorphlineSolrSink
a1.sinks.k1.morphlineFile = /root/aflume/conf/morphline.conf

#sink to avro
a1.sinks.k3.type = avro
a1.sinks.k3.channel = ch1
a1.sinks.k3.hostname = 192.168.0.135
a1.sinks.k3.port = 4545

# Finally, now that we've defined all of our components, tell
# agent1 which ones we want to activate.
a1.channels = ch1
a1.sources = mysqlbinlog
a1.sinks =  k3

　　flume.conf in host b

# Define a memory channel called ch1 on agent1
a1.channels.ch1.type = memory
#source avro
a1.sources.s1.channels = ch1
a1.sources.s1.type = avro
a1.sources.s1.bind = 0.0.0.0
a1.sources.s1.port = 4545
#sink to hdfs
a1.sinks.k1.channel = ch1
a1.sinks.k1.type = hdfs
a1.sinks.k1.hdfs.path = /user/flume/mysqlbinlog/%{tablename}
a1.sinks.k1.hdfs.filePrefix = %{tablename}-
a1.sinks.k1.hdfs.round = false
a1.sinks.k1.hdfs.rollInterval = 0
a1.sinks.k1.hdfs.rollCount = 0
a1.sinks.k1.hdfs.rollSize = 0
a1.sinks.k1.hdfs.batchSize = 100
a1.sinks.k1.hdfs.fileType = DataStream
a1.sinks.k1.hdfs.roundValue = 10
a1.sinks.k1.hdfs.roundUnit = minute
a1.sinks.k1.hdfs.useLocalTimeStamp = true

# Finally, now that we've defined all of our components, tell
# agent1 which ones we want to activate.
a1.channels = ch1
a1.sources = s1
a1.sinks = k1
　　There is a error
　　

 
　　－－－－－－－－－－－－－－－－－－－－
　　1. install    flume in the same machine with solr
　　2. add zookeeper-3.4.6.jar to flume/lib dir
　　3. put plugin in plugins.d dir

plugins.d/
└── mysqlbinlog
├── lib
│   └── mysqlbinlog-sources.jar
└── libext
└── open-replicator-1.0.7.jar
　　4. set mysql to master/salve replication set and start up master and slave mysql daemons
　　5. start up solr
　　6. test by use command

flume-ng agent -c ./conf/ -f conf/flume.conf -Dflume.root.logger=DEBUG,console -n a1
　　References
　　http://flume.apache.org/FlumeUserGuide.html#morphlinesolrsink
　　http://cloudera.github.io/cdk/docs/current/cdk-morphlines/index.html
　　http://cloudera.github.io/cdk/docs/current/cdk-morphlines/morphlinesReferenceGuide.html
　　http://www.cloudera.com/content/cloudera/en/documentation/core/latest/topics/search_flume_morphline_solr_sink_config_options.html
　　http://www.slideshare.net/cloudera/using-morphlines-for-onthefly-etl
　　http://kitesdk.org/docs/current/morphlines/morphlines-reference-guide.html#Implementing_your_own_Custom_Command
　　http://kitesdk.org/docs/current/morphlines/morphlines-reference-guide.html