纪录:Solr6.4.2+Flume1.7.0+morphline+kafka集成

kafka2solr.sources = source_from_kafka
kafka2solr.channels = customer_doc_channel
kafka2solr.sinks = solr_sink1

# For each one of the sources, the type is defined  
kafka2solr.sources.source_from_kafka.type = org.apache.flume.source.kafka.KafkaSource
kafka2solr.sources.source_from_kafka.channels = customer_doc_channel
kafka2solr.sources.source_from_kafka.batchSize = 100
kafka2solr.sources.source_from_kafka.useFlumeEventFormat=false
kafka2solr.sources.source_from_kafka.kafka.bootstrap.servers= kafkanode0:9092,kafkanode1:9092,kafkanode2:9092
kafka2solr.sources.source_from_kafka.kafka.topics = tablecardLogin
kafka2solr.sources.source_from_kafka.kafka.consumer.group.id = catering_customer_core_070327
kafka2solr.sources.source_from_kafka.kafka.consumer.auto.offset.reset=earliest

# Other config values specific to each type of channel(sink or source)  
# can be defined as well  
kafka2solr.channels.customer_doc_channel.type = file
kafka2solr.channels.customer_doc_channel.capacity=10000000
kafka2solr.channels.customer_doc_channel.checkpointDir = /home/arli/data/flume-ng/customer_doc/checkpoint
kafka2solr.channels.customer_doc_channel.dataDirs = /home/arli/data/flume-ng/customer_doc/data

kafka2solr.sinks.solr_sink1.type = org.apache.flume.sink.solr.morphline.MorphlineSolrSink
kafka2solr.sinks.solr_sink1.channel = customer_doc_channel
kafka2solr.sinks.solr_sink1.batchSize = 5000
kafka2solr.sinks.solr_sink1.batchDurationMillis = 2000
kafka2solr.sinks.solr_sink1.morphlineFile = /home/arli/flume-config/morphlines.conf
kafka2solr.sinks.solr_sink1.morphlineId=morphline1
kafka2solr.sinks.solr_sink1.isIgnoringRecoverableExceptions=true
#kafka2solr.sinks.solr_sink1.isProductionMode=true

#默认的内存是不够的。需要扩大内存。
export JAVA_OPTS="-Xms100m -Xmx500m -Dcom.sun.management.jmxremote"

#Flume官方下载的包少了一些Solr相关的包，需要把solr的lib目录加到flume的classpath下。
FLUME_CLASSPATH="/xxx/solr-6.4.2/contrib/morphlines-core/lib/*:/xxxi/solr-6.4.2/dist/*:/xxx/solr-6.4.2/dist/solrj-lib/*:/xxx/solr-6.4.2/server/solr-webapp/webapp/WEB-INF/lib/*"

log4j.logger.org.apache.flume.sink.solr=DEBUG
log4j.logger.org.kitesdk.morphline=TRACE

SOLR_LOCATOR : {
   solrUrl : "http:\/\/localhost:8983\/solr\/catering_customer_core1"
   solrHomeDir : "/xxx/server/solr/catering_customer_core1/conf"
}

morphlines : [
  {
    #customer morphline
    id : morphline1
    
    # Import all morphline commands in these java packages and their subpackages.
    # Other commands that may be present on the classpath are not visible to this morphline.
    importCommands : ["org.kitesdk.**", "org.apache.solr.**"]
    
    commands : [                 
    { 
        readJson {}
    } 
    
    {
        tryRules 
        {
            catchExceptions : false
            throwExceptionIfAllRulesFailed : true
            rules : [
            {
              commands : [
                { 
                    contains {topic : [tablecardLogin] } 
                }
                
                #field need to be indexed from json.
                {
                    extractJsonPaths {
                      flatten : false
                      paths : {
                        account:/account         
                        customer_id:/customerId
                        history_signin_dates:/opt_time
                        history_signin_timestamps:/opt_time
                        name:/name
                        sex:/sex
                        }
                    }
                }
              ]
              
            }
            

            # if desired, the last rule can serve as a fallback mechanism 
            # for records that don't match any rule:
            {
                    commands : [
                    { logWarn { format : "Ignoring record with unsupported input format: {}", args : ["@{}"] } }
                    { dropRecord {} }    
                    ]
            }
          ]
        }
    }

    {
        convertTimestamp {
            field : history_signin_dates
            inputFormats : ["yyyy-MM-dd HH:mm:ss"]
            inputTimezone : Asia/Shanghai
            outputFormat : "yyyy-MM-dd'T'HH:mm:ss'Z/SECOND'"
            outputTimezone : Asia/Shanghai
            }
    }

    {
        convertTimestamp {
            field : history_signin_timestamps
            inputFormats : ["yyyy-MM-dd HH:mm:ss"]
            inputTimezone : Asia/Shanghai
            outputFormat : "unixTimeInMillis"
            outputTimezone : UTC
            }
    }

    {
        java {
            imports : "import java.util.*;import org.kitesdk.morphline.api.Command;import org.kitesdk.morphline.api.Record;"
            code:     """
                            Object customerId = record.getFirstValue("customer_id");
                            Object account = record.getFirstValue("account");
                            record.put("id", account + "@" + customerId);
                            return child.process(record);
                    """
            }
    }
    
    {sanitizeUnknownSolrFields {solrLocator : ${SOLR_LOCATOR}}}

    #将数据导入到solr中
    {loadSolr {solrLocator : ${SOLR_LOCATOR}}}

     ]
  }
]

cp managed-schema schema.xml

bin/flume-ng agent --conf ~/flume-config/ -f ~/flume-config/flume.conf  -n kafka2solr &