flink-user mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Piotr Nowojski <pi...@data-artisans.com>
Subject Re: Write each group to its own file
Date Mon, 23 Oct 2017 19:35:02 GMT
You’re welcome :)

> On 23 Oct 2017, at 20:43, Rodrigo Lazoti <rodrigolazoti@gmail.com> wrote:
> 
> Piotr,
> 
> I did as you suggested and it worked perfectly.
> Thank you! :)
> 
> Best,
> Rodrigo
> 
> On Thu, Oct 12, 2017 at 5:11 AM, Piotr Nowojski <piotr@data-artisans.com <mailto:piotr@data-artisans.com>>
wrote:
> Hi,
> 
> There is no straightforward way to do that. First of all, the error you are getting is
because you are trying to start new application ( env.fromElements(items) ) inside your reduce
function.
> 
> To do what you want, you have to hash partition the products based on category (instead
of grouping by and reducing) and after that either:
> 
> 1. Sort the hash partitioned products and implement custom OutputFormat (maybe based
on FileOutputFormat), that would start a new file when key value has changed.
> 
> Or
> 
> 2. Implement custom OutputFormat (maybe based on FileOutputFormat), that would keep multiple
opened files - one file per category - and write records accordingly.
> 
> Note that both options require first to hash partition the products. 1. Will be more
CPU and memory consuming (have to sort the data), 2. Can exceed the maximum number of simultaneously
opened file if number of categories is very high.
> 
> Piotrek
> 
> > On 11 Oct 2017, at 17:47, rlazoti <rodrigolazoti@gmail.com <mailto:rodrigolazoti@gmail.com>>
wrote:
> >
> > Hi,
> >
> > Is there a way to write each group to its own file using the Dataset api
> > (Batch)?
> >
> > For example, lets use the following class:
> >
> > case class Product(name: String, category: String)
> >
> > And the following Dataset:
> >
> > val products = env.fromElements(Product("i7", "cpu"), Product("R5", "cpu"),
> > Product("gtx1080", "gpu"), Product("vega64", "gpu"), Product("evo250gb",
> > "ssd"))
> >
> > So in this example my output should be these 3 files:
> >
> > - cpu.csv
> > i7, cpu
> > R5, cpu
> >
> > - gpu.csv
> > gtx1080, gpu
> > vega64, gpu
> >
> > - ssd.csv
> > evo250gb, ssd
> >
> >
> > I tried the following code, but got
> > org.apache.flink.api.common.InvalidProgramException: Task not serializable.
> >
> > products.groupBy("category").reduceGroup { group: Iterator[Product] =>
> >  val items = group.toSeq
> >  env.fromElements(items).writeAsCsv(s"${items.head.category}.csv")
> >  items
> > }
> >
> > I welcome any of your inputs.
> >
> > Thanks!
> >
> >
> >
> > --
> > Sent from: http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/
<http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/>
> 
> 


Mime
View raw message