Return-Path: <user-return-7615-archive-asf-public=cust-asf.ponee.io@flink.apache.org>
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 0E0B3200B32
	for <archive-asf-public-internal@cust-asf2.ponee.io>; Thu, 23 Jun 2016 15:13:40 +0200 (CEST)
Received: by cust-asf.ponee.io (Postfix)
	id 0C68D160A59; Thu, 23 Jun 2016 13:13:40 +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 54C76160A58
	for <archive-asf-public@cust-asf.ponee.io>; Thu, 23 Jun 2016 15:13:39 +0200 (CEST)
Received: (qmail 24620 invoked by uid 500); 23 Jun 2016 13:13:38 -0000
Mailing-List: contact user-help@flink.apache.org; run by ezmlm
Precedence: bulk
List-Help: <mailto:user-help@flink.apache.org>
List-Unsubscribe: <mailto:user-unsubscribe@flink.apache.org>
List-Post: <mailto:user@flink.apache.org>
List-Id: <user.flink.apache.org>
Reply-To: user@flink.apache.org
Delivered-To: mailing list user@flink.apache.org
Received: (qmail 24610 invoked by uid 99); 23 Jun 2016 13:13:38 -0000
Received: from pnap-us-west-generic-nat.apache.org (HELO spamd4-us-west.apache.org) (209.188.14.142)
    by apache.org (qpsmtpd/0.29) with ESMTP; Thu, 23 Jun 2016 13:13:38 +0000
Received: from localhost (localhost [127.0.0.1])
	by spamd4-us-west.apache.org (ASF Mail Server at spamd4-us-west.apache.org) with ESMTP id 0ADEAC0C99
	for <user@flink.apache.org>; Thu, 23 Jun 2016 13:13:38 +0000 (UTC)
X-Virus-Scanned: Debian amavisd-new at spamd4-us-west.apache.org
X-Spam-Flag: NO
X-Spam-Score: 3.486
X-Spam-Level: ***
X-Spam-Status: No, score=3.486 tagged_above=-999 required=6.31
	tests=[DKIM_ADSP_CUSTOM_MED=0.001, NML_ADSP_CUSTOM_MED=1.2,
	RCVD_IN_DNSWL_NONE=-0.0001, SPF_SOFTFAIL=0.972, URI_HEX=1.313]
	autolearn=disabled
Received: from mx2-lw-eu.apache.org ([10.40.0.8])
	by localhost (spamd4-us-west.apache.org [10.40.0.11]) (amavisd-new, port 10024)
	with ESMTP id Ky8TXrdT7FmR for <user@flink.apache.org>;
	Thu, 23 Jun 2016 13:13:36 +0000 (UTC)
Received: from mwork.nabble.com (mwork.nabble.com [162.253.133.43])
	by mx2-lw-eu.apache.org (ASF Mail Server at mx2-lw-eu.apache.org) with ESMTP id A62D05F4E8
	for <user@flink.apache.org>; Thu, 23 Jun 2016 13:13:35 +0000 (UTC)
Received: from mjoe.nabble.com (unknown [162.253.133.57])
	by mwork.nabble.com (Postfix) with ESMTP id B8353C872F64
	for <user@flink.apache.org>; Thu, 23 Jun 2016 06:13:34 -0700 (MST)
Date: Thu, 23 Jun 2016 05:33:20 -0700 (PDT)
From: Sendoh <unicorn.banachi@gmail.com>
To: user@flink.apache.org
Message-ID: <1466685200481-7673.post@n4.nabble.com>
Subject: Iterate several kafka topics using the kafka connector
MIME-Version: 1.0
Content-Type: text/plain; charset=us-ascii
Content-Transfer-Encoding: 7bit
archived-at: Thu, 23 Jun 2016 13:13:40 -0000

Hi Flink developers,

Can I ask how could we iterate several Kafka topics using the Kafka
connector?

Our idea is like the following example:

List<DataStream&lt;JSONObject>> streams = new ArrayList<>();

// Iterate kafka topics
Iterator<String> topicIter = topicList.iterator();

        while (topicIter.hasNext()){

            String topic = topicIter.next();

            streams.add(env.addSource(new FlinkKafkaConsumer09<>(topic,
                    new JSONSchema(), properties)).rebalance());

        }

Our goal is to union several kafka data streams into one, given the topics
as a list:

        Iterator<DataStream&lt;JSONObject>> streamsIt = streams.iterator();

        DataStream<JSONObject> currentStream = streamsIt.next();
        while(streamsIt.hasNext()){
            DataStream<JSONObject> nextStream = streamsIt.next();
            currentStream = currentStream.union(nextStream);
        }

Cheers,

Sendoh



--
View this message in context: http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Iterate-several-kafka-topics-using-the-kafka-connector-tp7673.html
Sent from the Apache Flink User Mailing List archive. mailing list archive at Nabble.com.