Mailing-List: contact user-help@flume.apache.org; run by ezmlm
Precedence: bulk
Reply-To: user@flume.apache.org
Received-SPF: pass (athena.apache.org: domain of allanfeid@gmail.com
 designates 209.85.215.47 as permitted sender)
MIME-Version: 1.0
In-Reply-To: <1371004623.32292.YahooMailNeo@web163902.mail.gq1.yahoo.com>
References: <1371004623.32292.YahooMailNeo@web163902.mail.gq1.yahoo.com>
Date: Wed, 12 Jun 2013 10:09:56 -0400
Message-ID: 
 <CAN94UWfY_mKmFN5rpYKht7euxs=ML2rrZUFFvCxQsGk1GW44ig@mail.gmail.com>
Subject: Re: ElasticSearchSink does not work
From: Allan Feid <allanfeid@gmail.com>
To: user@flume.apache.org, shushuai zhu <sszhu@yahoo.com>
Content-Type: multipart/alternative; boundary=089e0158ca68884ee404def59194

--089e0158ca68884ee404def59194
Content-Type: text/plain; charset=ISO-8859-1

Hi Shushuai,

I've had a similar issue, and in my case it was because I was using the
same channel for multiple sinks. I believe what happens is whatever sink
can remove the event from the queue first will have it written out, but I
don't know the specifics since I haven't had a chance to read through the
codebase. If you add a second channel for your elasticsearch sink and make
sure your avro-source writes to both channels, you should see data going to
both locations.

Thanks,
Allan


On Tue, Jun 11, 2013 at 10:37 PM, shushuai zhu <sszhu@yahoo.com> wrote:

> Hi,
>
> I am new to Flume. I am trying to send data using Flume Client perl API to
> Flume Avro source then ElasticSearchSink to an ElasticSearch engine. I
> could make the file_roll sink to work by sending the data to some file.
> However, I am encountering issue with ElasticSearchSink. The data did not
> go through to ElasticSearch engine:
>
> use Flume::Client;
> my $ng_client = Flume::Client::Transceiver::Socket->new(host => 'host
> name', port => 41414);
> my $ng_requestor = Flume::Client::Requestor::FlumeNG->new(client =>
> $ng_client);
> my ($result, $response) = $ng_requestor->request('appendBatch', [{ headers
> => {}, body => "hello, this is sent from perl (using FlumeNG)"}]);
> print "$response\n";    # response will be 'OK' on success
>
> since the returned $response is not defined (again this worked when
> file_roll sink was used).
>
> The ElasticSearch engine is working since I could load data to it via
> LogStash's EalsticSearch output type.
>
> The Flume agent was installed by downloading tarball from Cloudera:
>
> http://archive.cloudera.com/cdh4/cdh/4/flume-ng-1.3.0-cdh4.3.0.tar.gz
>
> The flume.conf is as followings. I played around the "hostNames" with full
> name, IP address, etc. Do not see output message in flume.log. Could
> someone let me know what could potentially cause the issue?
>
> Thanks.
>
> Shushuai
>
>
>
> # Define a memory channel called ch1 on agent1
> agent1.channels = ch1
> agent1.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.
> agent1.sources = avro-source1
> agent1.sources.avro-source1.channels = ch1
> agent1.sources.avro-source1.type = avro
> agent1.sources.avro-source1.bind = 0.0.0.0
> agent1.sources.avro-source1.port = 41414
>
> # Define a local file sink that simply logs all events it receives (this
> works well)
> #agent1.sinks = localout
> #agent1.sinks.localout.type = file_roll
> #agent1.sinks.localout.sink.directory = /scratch/flume-ng/log
> #agent1.sinks.localout.sink.rollInterval = 0
> #agent1.sinks.localout.channel = ch1
>
> # Define ElasticSearchSink sink (this does not work)
> agent1.sinks = k1
> agent1.sinks.k1.type =
> org.apache.flume.sink.elasticsearch.ElasticSearchSink
> agent1.sinks.k1.hostNames = localhost:9300
> agent1.sinks.k1.indexName = flume
> agent1.sinks.k1.indexType = logs
> agent1.sinks.k1.clusterName = elasticsearch
> agent1.sinks.k1.batchSize = 2
> agent1.sinks.k1.serializer =
> org.apache.flume.sink.elasticsearch.ElasticSearchDynamicSerializer
> agent1.sinks.k1.channel = ch1
>

--089e0158ca68884ee404def59194
Content-Type: text/html; charset=ISO-8859-1
Content-Transfer-Encoding: quoted-printable

<div dir=3D"ltr">Hi Shushuai,<div><br></div><div style>I&#39;ve had a simil=
ar issue, and in my case it was because I was using the same channel for mu=
ltiple sinks. I believe what happens is whatever sink can remove the event =
from the queue first will have it written out, but I don&#39;t know the spe=
cifics since I haven&#39;t had a chance to read through the codebase. If yo=
u add a second channel for your elasticsearch sink and make sure your avro-=
source writes to both channels, you should see data going to both locations=
.</div>
<div style><br></div><div style>Thanks,</div><div style>Allan</div></div><d=
iv class=3D"gmail_extra"><br><br><div class=3D"gmail_quote">On Tue, Jun 11,=
 2013 at 10:37 PM, shushuai zhu <span dir=3D"ltr">&lt;<a href=3D"mailto:ssz=
hu@yahoo.com" target=3D"_blank">sszhu@yahoo.com</a>&gt;</span> wrote:<br>
<blockquote class=3D"gmail_quote" style=3D"margin:0 0 0 .8ex;border-left:1p=
x #ccc solid;padding-left:1ex"><div><div style=3D"font-size:12pt;font-famil=
y:times new roman,new york,times,serif"><div>Hi,</div><div>=A0</div><div>I =
am new=A0to Flume. I am trying to send data=A0using Flume Client perl API t=
o Flume Avro source then ElasticSearchSink to an ElasticSearch engine. I co=
uld make the file_roll sink to work by sending the data to some file. Howev=
er, I am encountering issue with ElasticSearchSink. The data did not go thr=
ough to ElasticSearch engine:</div>
<div>=A0</div><div>use Flume::Client;</div><div>my $ng_client =3D Flume::Cl=
ient::Transceiver::Socket-&gt;new(host =3D&gt; &#39;host name&#39;, port =
=3D&gt; 41414);<br>my $ng_requestor =3D Flume::Client::Requestor::FlumeNG-&=
gt;new(client =3D&gt; $ng_client);<br>
my ($result, $response) =3D $ng_requestor-&gt;request(&#39;appendBatch&#39;=
, [{ headers =3D&gt; {}, body =3D&gt; &quot;hello, this is sent from perl (=
using FlumeNG)&quot;}]);<br>print &quot;$response\n&quot;;=A0=A0=A0 # respo=
nse will be &#39;OK&#39;
 on success</div><div>=A0</div><div>since the returned $response is not def=
ined (again this worked when file_roll sink was used).</div><div>=A0</div><=
div>The ElasticSearch engine is working since I could load data to it via L=
ogStash&#39;s EalsticSearch output type.</div>
<div>=A0</div><div>The Flume agent was installed by downloading tarball fro=
m Cloudera:</div><div>=A0</div><div><a href=3D"http://archive.cloudera.com/=
cdh4/cdh/4/flume-ng-1.3.0-cdh4.3.0.tar.gz" target=3D"_blank">http://archive=
.cloudera.com/cdh4/cdh/4/flume-ng-1.3.0-cdh4.3.0.tar.gz</a></div>
<div>=A0</div><div>The flume.conf is as followings. I played around the &qu=
ot;hostNames&quot; with full name,=A0IP address, etc.=A0Do not see output m=
essage=A0in flume.log. Could someone let me know what could potentially cau=
se the issue?</div>
<div>=A0</div><div>Thanks.</div><div><var></var>=A0</div><div>Shushuai</div=
><div>=A0</div><div>=A0</div><div>=A0</div><div># Define a
 memory channel called ch1 on agent1<br>agent1.channels =3D ch1<br>agent1.c=
hannels.ch1.type =3D memory</div><div>=A0</div><div># Define an Avro source=
 called avro-source1 on agent1 and tell it to bind to <a href=3D"http://0.0=
.0.0:41414" target=3D"_blank">0.0.0.0:41414</a>. Connect it to channel ch1.=
<br>
agent1.sources =3D avro-source1<br>agent1.sources.avro-source1.channels =3D=
 ch1<br>agent1.sources.avro-source1.type =3D avro<br>agent1.sources.avro-so=
urce1.bind =3D 0.0.0.0<br>agent1.sources.avro-source1.port =3D 41414</div><=
div>=A0</div>
<div># Define a local file sink that simply logs all events it receives (th=
is works well)<br>#agent1.sinks =3D localout<br>#agent1.sinks.localout.type=
 =3D file_roll<br>#agent1.sinks.localout.sink.directory =3D /scratch/flume-=
ng/log<br>
#agent1.sinks.localout.sink.rollInterval =3D 0<br>#agent1.sinks.localout.ch=
annel =3D ch1</div><div>=A0</div><div># Define ElasticSearchSink sink (this=
 does not work)</div><div>agent1.sinks =3D k1<br>agent1.sinks.k1.type =3D
 org.apache.flume.sink.elasticsearch.ElasticSearchSink<br>agent1.sinks.k1.h=
ostNames =3D localhost:9300<br>agent1.sinks.k1.indexName =3D flume<br>agent=
1.sinks.k1.indexType =3D logs<br>agent1.sinks.k1.clusterName =3D elasticsea=
rch<br>
agent1.sinks.k1.batchSize =3D 2<br>agent1.sinks.k1.serializer =3D org.apach=
e.flume.sink.elasticsearch.ElasticSearchDynamicSerializer<br>agent1.sinks.k=
1.channel =3D ch1<br></div></div></div></blockquote></div><br></div>

--089e0158ca68884ee404def59194--