Return-Path: X-Original-To: archive-asf-public-internal@cust-asf2.ponee.io Delivered-To: archive-asf-public-internal@cust-asf2.ponee.io Received: from cust-asf.ponee.io (cust-asf.ponee.io [163.172.22.183]) by cust-asf2.ponee.io (Postfix) with ESMTP id D6C7F200BC3 for ; Fri, 18 Nov 2016 08:26:56 +0100 (CET) Received: by cust-asf.ponee.io (Postfix) id D55DA160B04; Fri, 18 Nov 2016 07:26:56 +0000 (UTC) Delivered-To: archive-asf-public@cust-asf.ponee.io Received: from mail.apache.org (hermes.apache.org [140.211.11.3]) by cust-asf.ponee.io (Postfix) with SMTP id 06CE5160AFE for ; Fri, 18 Nov 2016 08:26:55 +0100 (CET) Received: (qmail 87460 invoked by uid 500); 18 Nov 2016 07:26:54 -0000 Mailing-List: contact user-help@flink.apache.org; run by ezmlm Precedence: bulk List-Help: List-Unsubscribe: List-Post: List-Id: Reply-To: user@flink.apache.org Delivered-To: mailing list user@flink.apache.org Received: (qmail 87450 invoked by uid 99); 18 Nov 2016 07:26:54 -0000 Received: from pnap-us-west-generic-nat.apache.org (HELO spamd2-us-west.apache.org) (209.188.14.142) by apache.org (qpsmtpd/0.29) with ESMTP; Fri, 18 Nov 2016 07:26:54 +0000 Received: from localhost (localhost [127.0.0.1]) by spamd2-us-west.apache.org (ASF Mail Server at spamd2-us-west.apache.org) with ESMTP id 51C051A0745 for ; Fri, 18 Nov 2016 07:26:54 +0000 (UTC) X-Virus-Scanned: Debian amavisd-new at spamd2-us-west.apache.org X-Spam-Flag: NO X-Spam-Score: 2.379 X-Spam-Level: ** X-Spam-Status: No, score=2.379 tagged_above=-999 required=6.31 tests=[DKIM_SIGNED=0.1, DKIM_VALID=-0.1, DKIM_VALID_AU=-0.1, HTML_MESSAGE=2, RCVD_IN_DNSWL_NONE=-0.0001, RCVD_IN_MSPIKE_H3=-0.01, RCVD_IN_MSPIKE_WL=-0.01, RCVD_IN_SORBS_SPAM=0.5, SPF_PASS=-0.001] autolearn=disabled Authentication-Results: spamd2-us-west.apache.org (amavisd-new); dkim=pass (2048-bit key) header.d=gmail.com Received: from mx1-lw-eu.apache.org ([10.40.0.8]) by localhost (spamd2-us-west.apache.org [10.40.0.9]) (amavisd-new, port 10024) with ESMTP id 8ukgXyMyBQbm for ; Fri, 18 Nov 2016 07:26:52 +0000 (UTC) Received: from mail-wm0-f44.google.com (mail-wm0-f44.google.com [74.125.82.44]) by mx1-lw-eu.apache.org (ASF Mail Server at mx1-lw-eu.apache.org) with ESMTPS id E1F965F27E for ; Fri, 18 Nov 2016 07:26:51 +0000 (UTC) Received: by mail-wm0-f44.google.com with SMTP id a197so19478080wmd.0 for ; Thu, 17 Nov 2016 23:26:51 -0800 (PST) DKIM-Signature: v=1; a=rsa-sha256; c=relaxed/relaxed; d=gmail.com; s=20120113; h=mime-version:in-reply-to:references:from:date:message-id:subject:to; bh=k+10HiuhHLhKPFabBrjo2c9oXyc6n25SLM8zoihYHZo=; b=x55qxg1QwbnhiLJh4zdQudLhIEO9UagsKFpo9NAQD8qKn+yA8uNldg5aL6OsMBCYbr Cski6AtbR2n32CL5VfQAtrZBoB/MwiAIFwJmmGTonw2NQJtEV8dsf03oL8SKwydukT2I b0tEdeEomnVFgEUuTCkZNKcwXygRj6JunACRyjK4vRIycug5rFt4UKjWaG8bx6XpaA+9 P8QlnxVV6eyKSZl2Nuqk/SvlaWQLmvG/YEbUdvLHPJDHefwhjJ22h368Ohd4yKwitHLj ZX6wv8CUxmJB7qkZPGs0tUve4IrHcNudNH8reMJvc2yIk/W62t75QpJd1YDK+ExVssxH zMgA== X-Google-DKIM-Signature: v=1; a=rsa-sha256; c=relaxed/relaxed; d=1e100.net; s=20130820; h=x-gm-message-state:mime-version:in-reply-to:references:from:date :message-id:subject:to; bh=k+10HiuhHLhKPFabBrjo2c9oXyc6n25SLM8zoihYHZo=; b=M2RNzz5mwBOhi2phCj/PLpKPcbcdou6Ddf98PO4yhO60qRLTerRwzV6h+h8LcbuH7L rfGgYPvCIfyv3mOYhlZEcKl7j0GgkBBxv121ifhRBj8WPPgLB4Y/zwRkqFySFfENVd6f lsThhaVpzcMMEnLZ+U+RShe6kNSADqt600AsNViMya2ApX1e2pDG02pa+hqnt2Q9VKAv dh8XfdVeKsYmJz3n308CSIW6E6kKMgLP945DgDXjA9/oTJyde/lQTVvKJI7jRRAoBQht gQxCqgwbnme2mGzVAC8C/ydOAbc1R82fD6qf4ui3NetBghyvacSeh2g5EN9Zf+UceS2g bfSA== X-Gm-Message-State: AKaTC002b8ZCh7CtagcF3eKQhiSgNxkJtFZjkMAQhfxaT8ojgoGkcJLFLxQLGpDNglGRp+nUj/1Psa47VWqV5Q== X-Received: by 10.28.63.3 with SMTP id m3mr8907107wma.113.1479454011377; Thu, 17 Nov 2016 23:26:51 -0800 (PST) MIME-Version: 1.0 Received: by 10.28.209.66 with HTTP; Thu, 17 Nov 2016 23:26:50 -0800 (PST) Received: by 10.28.209.66 with HTTP; Thu, 17 Nov 2016 23:26:50 -0800 (PST) In-Reply-To: References: From: CPC Date: Fri, 18 Nov 2016 10:26:50 +0300 Message-ID: Subject: Re: spark vs flink batch performance To: user@flink.apache.org Content-Type: multipart/alternative; boundary=001a114b7e1ecf4ef405418e3aca archived-at: Fri, 18 Nov 2016 07:26:57 -0000 --001a114b7e1ecf4ef405418e3aca Content-Type: text/plain; charset=UTF-8 Hi all, In the mean time i have three workers. Any thoughts about improving flink performance? Thank you... On Nov 17, 2016 00:38, "CPC" wrote: > Hi all, > > I am trying to compare spark and flink batch performance. In my test i am > using ratings.csv in http://files.grouplens.org/ > datasets/movielens/ml-latest.zip dataset. I also concatenated ratings.csv > 16 times to increase dataset size(total of 390465536 records almost 10gb).I > am reading from google storage with gcs-connector and file schema is : > userId,movieId,rating,timestamp. Basically i am calculating average > rating per movie > > Code for flink(i tested CombineHint.HASH and CombineHint.SORT) > > case class Rating(userID: String, movieID: String, rating: Double, date: >> Timestamp) >> > > >> def parseRating(line: String): Rating = { >> val arr = line.split(",") >> Rating(arr(0), arr(1), arr(2).toDouble, new Timestamp((arr(3).toLong * >> 1000))) >> } > > > > val ratings: DataSet[Rating] = env.readTextFile("gs://cpcflink/wikistream/ratingsheadless16x.csv").map(a >> => parseRating(a)) >> ratings >> .map(i => (i.movieID, 1, i.rating)) >> .groupBy(0).reduce((l, r) => (l._1, l._2 + r._2, l._3 + r._3), >> CombineHint.HASH) >> .map(i => (i._1, i._3 / i._2)).collect().sortBy(_._1). >> sortBy(_._2)(Ordering.Double.reverse).take(10) > > > with CombineHint.HASH 3m49s and with CombineHint.SORT 5m9s > > Code for Spark(i tested reduceByKey and reduceByKeyLocaly) > >> case class Rating(userID: String, movieID: String, rating: Double, date: >> Timestamp) >> def parseRating(line: String): Rating = { >> val arr = line.split(",") >> Rating(arr(0), arr(1), arr(2).toDouble, new Timestamp((arr(3).toLong * >> 1000))) >> } >> val conf = new SparkConf().setAppName("Simple Application") >> val sc = new SparkContext(conf) >> val keyed: RDD[(String, (Int, Double))] = sc.textFile("gs://cpcflink/ >> wikistream/ratingsheadless16x.csv").map(parseRating).map(r => >> (r.movieID, (1, r.rating))) >> keyed.reduceByKey((l, r) => (l._1 + r._1, l._2 + r._2)).mapValues(i => >> i._2 / i._1).collect.sortBy(_._1).sortBy(a=>a._2)(Ordering. >> Double.reverse).take(10).foreach(println) > > > with reduceByKeyLocaly 2.9 minute(almost 2m54s) and reduceByKey 3.1 > minute(almost 3m6s) > > Machine config on google cloud: > taskmanager/sparkmaster: n1-standard-1 (1 vCPU, 3.75 GB memory) > jobmanager/sparkworkers: n1-standard-2 (2 vCPUs, 7.5 GB memory) > java version:jdk jdk-8u102 > flink:1.1.3 > spark:2.0.2 > > I also attached flink-conf.yaml. Although it is not such a big difference > there is a 40% performance difference between spark and flink. Is there > something i am doing wrong? If there is not how can i fine tune flink or is > it normal spark has better performance with batch data? > > Thank you in advance... > --001a114b7e1ecf4ef405418e3aca Content-Type: text/html; charset=UTF-8 Content-Transfer-Encoding: quoted-printable

Hi all,

In the mean time i have three workers. Any thoughts about im= proving flink performance?

Thank you...


On Nov 17, 2016 0= 0:38, "CPC" <achalil@gmai= l.com> wrote:
case c= lass Rating(userID: String, movieID: String, rating: Double, date: Timestam= p)
def parseRating(line: String): Rating =3D {
=C2=A0 val arr = =3D line.split(",")
=C2=A0 Rating(arr(0), arr(1), arr(2).toDou= ble, new Timestamp((arr(3).toLong * 1000)))
}
--001a114b7e1ecf4ef405418e3aca--