@ Override public void setup ( JavaStreamingContext jssc , CommandLine cli ) throws Exception { String filtersArg = cli . getOptionValue ( "tweetFilters" ) ; String [ ] filters = ( filtersArg != null ) ? filtersArg . split ( "," ) : new String [ 0 ] ; // start receiving a stream of tweets ... JavaReceiverInputDStream < Status > tweets = TwitterUtils . createStream ( jssc , null , filters ) ; String fusionUrl = cli . getOptionValue ( "fusion" ) ; if ( fusionUrl != null ) { // just send JSON directly to Fusion SolrSupport . sendDStreamOfDocsToFusion ( fusionUrl , cli . getOptionValue ( "fusionCredentials" ) , tweets . dstream ( ) , batchSize ) ; } else { // map incoming tweets into PipelineDocument objects for indexing in Solr JavaDStream < SolrInputDocument > docs = tweets . map ( new Function < Status , SolrInputDocument > ( ) { /** * Convert a twitter4j Status object into a SolrJ SolrInputDocument */ public SolrInputDocument call ( Status status ) { if ( log . isDebugEnabled ( ) ) { log . debug ( "Received tweet: " + status . getId ( ) + ": " + status . getText ( ) . replaceAll ( "\\s+" , " " ) ) ; } // simple mapping from primitives to dynamic Solr fields using reflection SolrInputDocument doc = SolrSupport . autoMapToSolrInputDoc ( "tweet-" + status . getId ( ) , status , null ) ; doc . setField ( "provider_s" , "twitter" ) ; doc . setField ( "author_s" , status . getUser ( ) . getScreenName ( ) ) ; doc . setField ( "type_s" , status . isRetweet ( ) ? "echo" : "post" ) ; if ( log . isDebugEnabled ( ) ) log . debug ( "Transformed document: " + doc . toString ( ) ) ; return doc ; } } ) ; // when ready, send the docs into a SolrCloud cluster SolrSupport . indexDStreamOfDocs ( zkHost , collection , batchSize , docs . dstream ( ) ) ; } }
Sends a stream of tweets to Solr .