Mailing-List: contact issues-help@spark.apache.org; run by ezmlm
Precedence: bulk
Date: Thu, 24 Mar 2016 07:13:25 +0000 (UTC)
From: "Ben Teeuwen (JIRA)" <jira@apache.org>
To: issues@spark.apache.org
Message-ID: <JIRA.12950945.1458152928000.38935.1458803605876@Atlassian.JIRA>
In-Reply-To: <JIRA.12950945.1458152928000@Atlassian.JIRA>
References: <JIRA.12950945.1458152928000@Atlassian.JIRA>
 <JIRA.12950945.1458152928342@arcas>
Subject: [jira] [Commented] (SPARK-13939) Kafka createDirectStream not
 parallelizing properly
MIME-Version: 1.0
Content-Type: text/plain; charset=utf-8
Content-Transfer-Encoding: quoted-printable


    [ https://issues.apache.org/jira/browse/SPARK-13939?page=3Dcom.atlassia=
n.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=3D152=
09901#comment-15209901 ]=20

Ben Teeuwen commented on SPARK-13939:
-------------------------------------

Thanks a lot Cody for narrowing the search space. The problem was the forma=
t of the data going into the kafka producer; tuples with an empty key and t=
he actual event as the value.

At http://stackoverflow.com/questions/26553412/produce-kafka-message-to-sel=
ected-partition, they mention "The partitioner class for partitioning messa=
ges amongst sub-topics. The default partitioner is based on the hash of the=
 key." Now I noticed, but never questioned, that incoming events from Kafka=
 have are tuples. The key is empty, the value contains the actual event. So=
 first thing I did in Spark was (see also pastes above);

{code}
  stream_it =3D (directKafkaStream
    .map(lambda (key, js_string): json.loads(js_string)) # raw event json b=
ecomes a dictionary; empty weird kafka key is ignored.)
{code}

The hash for that empty key produced by the kafka-console-producer will be =
the same, so it is sent to 1 partition. We changed the key to a random numb=
er and now it works fluently :).

> Kafka createDirectStream not parallelizing properly
> ---------------------------------------------------
>
>                 Key: SPARK-13939
>                 URL: https://issues.apache.org/jira/browse/SPARK-13939
>             Project: Spark
>          Issue Type: Bug
>          Components: Streaming
>            Reporter: Ben Teeuwen
>         Attachments: 0CEAF9A0-6637-44BB-95B2-2529992723A2.png, 215B28E2-6=
38B-494C-8084-FD46E9984522.png, 4E119936-14E3-490E-A885-7D2E2CB2940F.png, 9=
F0FF528-85DF-475D-9507-8FBF93C46750.png, ECBE2DFF-6B35-48C5-B692-B9A80FC1E3=
F5.png, screenshot-1.png
>
>
> I=E2=80=99m trying to get a streaming app running using pyspark (1.6.0), =
Kafka and the receiverless direct approach =E2=80=98createDirectStream=E2=
=80=99. But it seemingly has problems with the degree of parallelism in Spa=
rk. I=E2=80=99ve written the app both in Scala and Pyspark; both exhibit th=
e same behavior.
> Context:
> - stream with 10-30k events per 10 seconds batch size.
> - kafka topic has 10 partitions.
> - createDirectStream with kafkaparams only metadata.broker.list, containi=
ng 4 brokers.
> - 10 executors 2 cores each, 3gb ram + 3gb ram driver mem.
> - backpressure on
> - not using speculative execution
> - simple logic: parse json, create key-value tuple, flatmap, reduceByKey,=
 pprint to screen. It is supposed to be keeping track of states, but for no=
w I'm unfortunately having issues with a simple printing of the minimum and=
 maximum epoch.
> At the start of the streaming (e.g. started just now at 19.07):
> First thing I do is repartition to spread the events evenly over all the =
executors. Looking at the streaming tab > batch details > Input Metadata, I=
 see it ingests only from 1 kafka partition:
> {code}
> Kafka direct stream [0]=09
>     topic: test    partition: 1    offsets: 16630012 to 16639226
> {code}
> One executor is doing the repartitioning, and is taking more than the bat=
ch interval time. So backpressure kicks in. The events ingested as trimmed =
down to a 100. That gets processed in 2 seconds. Then slowly, more Kafka pa=
rtitions are being used. E.g. 10 minutes later:
> {code}
> Kafka direct stream [0]=09
>     topic: test    partition: 9    offsets: 16262300 to 16262400
>     topic: test    partition: 1    offsets: 16683171 to 16683271
> {code}
> When running for a day, the amount of kafka partitions it ingests from st=
abilizes around 3-6 partitions. But it never ingests the full stream, thoug=
h it has more partitions to ingest from in parallel and executors to utiliz=
e. E.g. half an hour later:
> {code}
> Kafka direct stream [0]=09
>     topic: test    partition: 9    offsets: 16327090 to 16328090
>     topic: test    partition: 6    offsets: 17140538 to 17141538
>     topic: test    partition: 0    offsets: 22776394 to 22777394
>     topic: test    partition: 1    offsets: 16747961 to 16748961
>     topic: test    partition: 7    offsets: 15090120 to 15091120
> {code}
> So it loses of a lot of events, and it processes older events in later ba=
tches. E.g. printing min/max timestamps shows very events going back almost=
 to the start of the streaming app. E.g.
> {code}
> #### Printing at 16-03-16 19:36:33
> ### min 16-03-16 19:09:12 (epoch =3D 1458151752)
> #### Printing at 16-03-16 19:36:34
> ### max 16-03-16 19:31:51 (epoch =3D 1458153111)
> #### Printing at 16-03-16 19:36:42
> ### min 16-03-16 19:09:12 (epoch =3D 1458151752)
> #### Printing at 16-03-16 19:36:43
> ### max 16-03-16 19:31:51 (epoch =3D 1458153111)
> {code}
> My take from the =E2=80=98Simplified Parallelism=E2=80=99 bullet in the d=
ocs (http://spark.apache.org/docs/latest/streaming-kafka-integration.html),=
 is not to worry about parallellism, as long as I provide sufficient resour=
ces. And 10 execs with 2 cores receiving from a kafka stream with 10 partit=
ions, containing 10-30k events per 10 seconds, seems plentiful.
> (this was discussed during Amsterdam Spark Meetup March 14 2016 with [~ho=
ldenk_amp], and she advised to write it up in a ticket here).


--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@spark.apache.org
For additional commands, e-mail: issues-help@spark.apache.org