Return-Path: <user-return-8871-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 9B6D9200B3E
	for <archive-asf-public-internal@cust-asf2.ponee.io>; Wed,  7 Sep 2016 12:18:50 +0200 (CEST)
Received: by cust-asf.ponee.io (Postfix)
	id 99F4A160AC1; Wed,  7 Sep 2016 10:18:50 +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 E0BB9160AA3
	for <archive-asf-public@cust-asf.ponee.io>; Wed,  7 Sep 2016 12:18:49 +0200 (CEST)
Received: (qmail 89770 invoked by uid 500); 7 Sep 2016 10:18:49 -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 89760 invoked by uid 99); 7 Sep 2016 10:18:48 -0000
Received: from pnap-us-west-generic-nat.apache.org (HELO spamd3-us-west.apache.org) (209.188.14.142)
    by apache.org (qpsmtpd/0.29) with ESMTP; Wed, 07 Sep 2016 10:18:48 +0000
Received: from localhost (localhost [127.0.0.1])
	by spamd3-us-west.apache.org (ASF Mail Server at spamd3-us-west.apache.org) with ESMTP id 94AD5180628
	for <user@flink.apache.org>; Wed,  7 Sep 2016 10:18:48 +0000 (UTC)
X-Virus-Scanned: Debian amavisd-new at spamd3-us-west.apache.org
X-Spam-Flag: NO
X-Spam-Score: 3.335
X-Spam-Level: ***
X-Spam-Status: No, score=3.335 tagged_above=-999 required=6.31
	tests=[FREEMAIL_ENVFROM_END_DIGIT=0.25, KAM_ASCII_DIVIDERS=0.8,
	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 (spamd3-us-west.apache.org [10.40.0.10]) (amavisd-new, port 10024)
	with ESMTP id 5397r8n7MQQx for <user@flink.apache.org>;
	Wed,  7 Sep 2016 10:18:45 +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 12D7F5FBD0
	for <user@flink.apache.org>; Wed,  7 Sep 2016 10:18:44 +0000 (UTC)
Received: from mjoe.nabble.com (unknown [162.253.133.57])
	by mwork.nabble.com (Postfix) with ESMTP id 44B8C49CB4695
	for <user@flink.apache.org>; Wed,  7 Sep 2016 03:18:43 -0700 (MST)
Date: Wed, 7 Sep 2016 03:15:39 -0700 (PDT)
From: jiecxy <253441801@qq.com>
To: user@flink.apache.org
Message-ID: <1473243339074-8944.post@n4.nabble.com>
Subject: How to assign timestamp for event time in a stream?
MIME-Version: 1.0
Content-Type: text/plain; charset=us-ascii
Content-Transfer-Encoding: 7bit
archived-at: Wed, 07 Sep 2016 10:18:50 -0000

The program is to read the unordered records from a log file, and to print
the record in order. But it doesn't change the  order, is there anything
wrong in my code?  Can anyone give me an example?


This is my program:

Note: the class Tokenizer is to transfer the log to four parts. Like this: 
   Sep  6 09:28:01 master systemd: Stopping user-988.slice. 
to 
  Tuple4<time, master, systemd,  Stopping user-988.slice.>

---------------------------------------------------------------------------------------------------------
        // set up the execution environment
        final StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        env.setParallelism(1);

        StreamTableEnvironment tableEnv =
TableEnvironment.getTableEnvironment(env);

        DataStream<String> text = env.addSource(new
FlinkKafkaConsumer08<String>("test", new SimpleStringSchema(), properties));
        DataStream<Tuple4&lt;Long, String, String, String>> messages =
                text.flatMap(new Tokenizer())
                        .assignTimestampsAndWatermarks(new
AscendingTimestampExtractor<Tuple4&lt;Long, String, String, String>>() {
                            @Override
                            public long
extractAscendingTimestamp(Tuple4<Long, String, String, String> tuple4) {
                                return tuple4.f0;
                            }
                        });
        messages.print().setParallelism(1);
---------------------------------------------------------------------------------------------------------

The inputs like this:
Sep  6 09:28:01 master systemd: Stopping user-988.slice1.
Sep  6 09:28:04 master systemd: Stopping user-988.slice4.
Sep  6 09:28:03 master systemd: Stopping user-988.slice3.
Sep  6 09:28:02 master systemd: Stopping user-988.slice2.

But the outputs are same with input, it doesn't change them to 1 - 2 - 3 - 4
(order by the time)...



--
View this message in context: http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/How-to-assign-timestamp-for-event-time-in-a-stream-tp8944.html
Sent from the Apache Flink User Mailing List archive. mailing list archive at Nabble.com.