You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@flink.apache.org by "harshit.varshney@iktara.ai" <Ha...@iktara.ai> on 2022/06/02 14:01:24 UTC
Issue Facing While Using EmbeddedRocksDbCheckpointing FlinkVersion(1.15.0)
Dear Team,
I am new to pyflink and request for your support in issue I am facing with
Pyflink. I am using Pyflink version 1.15.0 & using reference code from
pyflink reference code.
The errors I am getting
Traceback (most recent call last):
File
"E:\pythonProject16\lib\site-packages\apache_beam\runners\worker\data_plane.
py", line 470, in input_elements
element = received.get(timeout=1)
File "C:\Users\Admin\AppData\Local\Programs\Python\Python38\lib\queue.py",
line 178, in get
raise Empty
_queue.Empty
RuntimeError: Channel closed prematurely.
My code is:
import json
import os
import time
from datetime import datetime
from pyflink.common import SimpleStringSchema, JsonRowDeserializationSchema,
Types, JsonRowSerializationSchema
from pyflink.datastream import StreamExecutionEnvironment, WindowFunction,
HashMapStateBackend, CheckpointingMode, \
FileSystemCheckpointStorage, KeyedProcessFunction, RuntimeContext,
EmbeddedRocksDBStateBackend, RocksDBStateBackend
from pyflink.datastream.connectors import FlinkKafkaConsumer,
FlinkKafkaProducer
from pyflink.datastream.state import ValueStateDescriptor, StateTtlConfig,
ListStateDescriptor
from sklearn.preprocessing import LabelEncoder
import pickle
import pandas as pd
from pyflink.common import Row
import argparse
from typing import Iterable
from pyflink.datastream.connectors import FileSink, OutputFileConfig,
RollingPolicy
from pyflink.common import Types, WatermarkStrategy, Time, Encoder
from pyflink.common.watermark_strategy import TimestampAssigner
from pyflink.datastream import StreamExecutionEnvironment,
ProcessWindowFunction
from pyflink.datastream.window import TumblingEventTimeWindows, TimeWindow,
TumblingProcessingTimeWindows
class MyTimestampAssigner(TimestampAssigner):
def extract_timestamp(self, value, record_timestamp) -> int:
return int(value[0])
class CountWindowFunction(WindowFunction[tuple, tuple, str, TimeWindow]):
def apply(self, key: str, window: TimeWindow, inputs: Iterable[tuple]):
# return [(key, result)]
return [(key, len([e for e in inputs]))]
class Storage(KeyedProcessFunction):
def __init__(self):
self.state = None
def open(self, runtime_context: RuntimeContext):
state_descriptor = ValueStateDescriptor("state", Types.FLOAT())
state_ttl_config = StateTtlConfig \
.new_builder(Time.days(7)) \
.set_update_type(StateTtlConfig.UpdateType.OnReadAndWrite) \
.disable_cleanup_in_background() \
.build()
state_descriptor.enable_time_to_live(state_ttl_config)
self.state = runtime_context.get_state(state_descriptor)
def process_element(self, value, ctx: 'KeyedProcessFunction.Context'):
# retrieve the current count
current = self.state.value()
if current is None:
current = 0
current = value[1]
self.state.update(current)
yield current,time.time()
def write_to_kafka():
env = StreamExecutionEnvironment.get_execution_environment()
env.set_parallelism(1)
env.enable_checkpointing(1000)
env.get_checkpoint_config().set_min_pause_between_checkpoints(5000)
env.set_state_backend(EmbeddedRocksDBStateBackend())
env.get_checkpoint_config().set_checkpointing_mode(CheckpointingMode.AT_LEAS
T_ONCE)
#env.get_checkpoint_config().enable_unaligned_checkpoints()
check = os.path.join(os.path.abspath(os.path.dirname(__file__)),
'checkpoint-dir11')
env.get_checkpoint_config().set_checkpoint_storage(FileSystemCheckpointStora
ge("file:///{}".format(check)))
kafka_jar = os.path.join(os.path.abspath(os.path.dirname(__file__)),
'flink-sql-connector-kafka_2.11-1.14.4.jar')
env.add_jars("file:///{}".format(kafka_jar))
rocksdb_jar = os.path.join(os.path.abspath(os.path.dirname(__file__)),
'flink-statebackend-rocksdb_2.11-1.0.0.jar')
env.add_jars("file:///{}".format(rocksdb_jar))
# deserialization_schema = SimpleStringSchema()
deserialization_schema = JsonRowDeserializationSchema.builder() \
.type_info(type_info=Types.ROW_NAMED(["time_stamp",
"Bill_number", "Store_Code",
"itemdescription", "Item_code",
"Gross_Price", "Discount",
"Net_Price",
"purchaseorReturn",
"Membership_No",
"Billing_Date", "Billing_Time"],
[Types.DOUBLE(),
Types.STRING(), Types.INT(), Types.STRING(),
Types.STRING(), Types.FLOAT(),
Types.FLOAT(), Types.FLOAT(),
Types.STRING(),
Types.STRING(), Types.STRING(),
Types.STRING()])).build()
kafka_consumer = FlinkKafkaConsumer(
topics='test',
deserialization_schema=deserialization_schema,
properties={'bootstrap.servers': '192.168.1.37:9092', 'group.id':
'test_group1'})
ds = env.add_source(kafka_consumer)
# ds.print()
watermark_strategy = WatermarkStrategy.for_monotonous_timestamps() \
.with_timestamp_assigner(MyTimestampAssigner())
# ds = ds.map(lambda x: eval(x))
ds1 = ds.assign_timestamps_and_watermarks(watermark_strategy) \
.key_by(lambda x: '/root', key_type=Types.STRING()) \
.window(TumblingProcessingTimeWindows.of(Time.seconds(1))) \
.apply(CountWindowFunction(), Types.TUPLE([Types.STRING(),
Types.INT()]))
ds1.print()
ds4 = ds1.key_by(lambda value: value[0]) \
.process(Storage())
ds4.print()
# ds1 = ds.map(mlfunc)
# ds1.print()
env.execute('write_to_kafka')
if __name__ == '__main__':
print("start writing data to kafka")
write_to_kafka()
Re: FW: FW: Issue Facing While Using EmbeddedRocksDbCheckpointing FlinkVersion(1.15.0)
Posted by Shuiqiang Chen <ac...@gmail.com>.
Hi，
From the task manager's log, We can find the following exception stack
trace, seems it was the operating system related problem with rocksDB.
2022-06-04 14:45:53,878 INFO
> org.apache.flink.runtime.executiongraph.ExecutionGraph [] - KEYED
> PROCESS, Map -> Sink: Print to Std. Out (1/1)
> (3a295a887361f4f87d2eaf79e901d056) switched from INITIALIZING to FAILED on
> 2f04093f-a18f-4c2c-b02e-6d1e6ece677e @ kubernetes.docker.internal
> (dataPort=-1).
> java.lang.Exception: Exception while creating StreamOperatorStateContext.
> at
> org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.streamOperatorStateContext(StreamTaskStateInitializerImpl.java:255)
> ~[flink-dist_2.11-1.14.4.jar:1.14.4]
> at
> org.apache.flink.streaming.api.operators.AbstractStreamOperator.initializeState(AbstractStreamOperator.java:268)
> ~[flink-dist_2.11-1.14.4.jar:1.14.4]
> at
> org.apache.flink.streaming.runtime.tasks.RegularOperatorChain.initializeStateAndOpenOperators(RegularOperatorChain.java:109)
> ~[flink-dist_2.11-1.14.4.jar:1.14.4]
> at
> org.apache.flink.streaming.runtime.tasks.StreamTask.restoreGates(StreamTask.java:711)
> ~[flink-dist_2.11-1.14.4.jar:1.14.4]
> at
> org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$1.call(StreamTaskActionExecutor.java:55)
> ~[flink-dist_2.11-1.14.4.jar:1.14.4]
> at
> org.apache.flink.streaming.runtime.tasks.StreamTask.restoreInternal(StreamTask.java:687)
> ~[flink-dist_2.11-1.14.4.jar:1.14.4]
> at
> org.apache.flink.streaming.runtime.tasks.StreamTask.restore(StreamTask.java:654)
> ~[flink-dist_2.11-1.14.4.jar:1.14.4]
> at
> org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:958)
> ~[flink-dist_2.11-1.14.4.jar:1.14.4]
> at
> org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:927)
> ~[flink-dist_2.11-1.14.4.jar:1.14.4]
> at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:766)
> ~[flink-dist_2.11-1.14.4.jar:1.14.4]
> at org.apache.flink.runtime.taskmanager.Task.run(Task.java:575)
> ~[flink-dist_2.11-1.14.4.jar:1.14.4]
> at java.lang.Thread.run(Thread.java:748) ~[?:1.8.0_202]
> Caused by: org.apache.flink.util.FlinkException: Could not restore keyed
> state backend for
> PythonKeyedProcessOperator_ce1331762e4644ef89dbdbc15321f049_(1/1) from any
> of the 1 provided restore options.
> at
> org.apache.flink.streaming.api.operators.BackendRestorerProcedure.createAndRestore(BackendRestorerProcedure.java:160)
> ~[flink-dist_2.11-1.14.4.jar:1.14.4]
> at
> org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.keyedStatedBackend(StreamTaskStateInitializerImpl.java:346)
> ~[flink-dist_2.11-1.14.4.jar:1.14.4]
> at
> org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.streamOperatorStateContext(StreamTaskStateInitializerImpl.java:164)
> ~[flink-dist_2.11-1.14.4.jar:1.14.4]
> ... 11 more
> Caused by: java.io.IOException: Could not load the native RocksDB library
> at
> org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend.ensureRocksDBIsLoaded(EmbeddedRocksDBStateBackend.java:882)
> ~[flink-dist_2.11-1.14.4.jar:1.14.4]
> at
> org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend.createKeyedStateBackend(EmbeddedRocksDBStateBackend.java:402)
> ~[flink-dist_2.11-1.14.4.jar:1.14.4]
> at
> org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend.createKeyedStateBackend(EmbeddedRocksDBStateBackend.java:93)
> ~[flink-dist_2.11-1.14.4.jar:1.14.4]
> at
> org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.lambda$keyedStatedBackend$1(StreamTaskStateInitializerImpl.java:329)
> ~[flink-dist_2.11-1.14.4.jar:1.14.4]
> at
> org.apache.flink.streaming.api.operators.BackendRestorerProcedure.attemptCreateAndRestore(BackendRestorerProcedure.java:168)
> ~[flink-dist_2.11-1.14.4.jar:1.14.4]
> at
> org.apache.flink.streaming.api.operators.BackendRestorerProcedure.createAndRestore(BackendRestorerProcedure.java:135)
> ~[flink-dist_2.11-1.14.4.jar:1.14.4]
> at
> org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.keyedStatedBackend(StreamTaskStateInitializerImpl.java:346)
> ~[flink-dist_2.11-1.14.4.jar:1.14.4]
> at
> org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.streamOperatorStateContext(StreamTaskStateInitializerImpl.java:164)
> ~[flink-dist_2.11-1.14.4.jar:1.14.4]
> ... 11 more
> Caused by: java.lang.UnsatisfiedLinkError:
> C:\Users\Admin\AppData\Local\Temp\rocksdb-lib-cfe23cb49ce486a76d1126db2517e679\librocksdbjni-win64.dll:
> A dynamic link library (DLL) initialization routine failed
> at java.lang.ClassLoader$NativeLibrary.load(Native Method) ~[?:1.8.0_202]
> at java.lang.ClassLoader.loadLibrary0(ClassLoader.java:1941) ~[?:1.8.0_202]
> at java.lang.ClassLoader.loadLibrary(ClassLoader.java:1824) ~[?:1.8.0_202]
> at java.lang.Runtime.load0(Runtime.java:809) ~[?:1.8.0_202]
> at java.lang.System.load(System.java:1086) ~[?:1.8.0_202]
> at
> org.rocksdb.NativeLibraryLoader.loadLibraryFromJar(NativeLibraryLoader.java:79)
> ~[flink-dist_2.11-1.14.4.jar:1.14.4]
> at
> org.rocksdb.NativeLibraryLoader.loadLibrary(NativeLibraryLoader.java:57)
> ~[flink-dist_2.11-1.14.4.jar:1.14.4]
> at
> org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend.ensureRocksDBIsLoaded(EmbeddedRocksDBStateBackend.java:856)
> ~[flink-dist_2.11-1.14.4.jar:1.14.4]
> at
> org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend.createKeyedStateBackend(EmbeddedRocksDBStateBackend.java:402)
> ~[flink-dist_2.11-1.14.4.jar:1.14.4]
> at
> org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend.createKeyedStateBackend(EmbeddedRocksDBStateBackend.java:93)
> ~[flink-dist_2.11-1.14.4.jar:1.14.4]
> at
> org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.lambda$keyedStatedBackend$1(StreamTaskStateInitializerImpl.java:329)
> ~[flink-dist_2.11-1.14.4.jar:1.14.4]
> at
> org.apache.flink.streaming.api.operators.BackendRestorerProcedure.attemptCreateAndRestore(BackendRestorerProcedure.java:168)
> ~[flink-dist_2.11-1.14.4.jar:1.14.4]
> at
> org.apache.flink.streaming.api.operators.BackendRestorerProcedure.createAndRestore(BackendRestorerProcedure.java:135)
> ~[flink-dist_2.11-1.14.4.jar:1.14.4]
> at
> org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.keyedStatedBackend(StreamTaskStateInitializerImpl.java:346)
> ~[flink-dist_2.11-1.14.4.jar:1.14.4]
> at
> org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.streamOperatorStateContext(StreamTaskStateInitializerImpl.java:164)
> ~[flink-dist_2.11-1.14.4.jar:1.14.4]
> ... 11 more
>
harshit.varshney@iktara.ai <Ha...@iktara.ai> 于2022年6月9日周四
15:06写道：
>
>
>
>
> *From:* Dian Fu [mailto:dian0511.fu@gmail.com]
> *Sent:* Monday, June 6, 2022 6:47 AM
> *To:* harshit.varshney@iktara.ai
> *Cc:* user
> *Subject:* Re: FW: Issue Facing While Using EmbeddedRocksDbCheckpointing
> FlinkVersion(1.15.0)
>
>
>
> None of the attachments are logs of the TaskManger. The TaskManger log
> should be located in the directory
> `E:\pythonProject16\lib\site-packages\pyflink\log`.
>
>
>
> On Fri, Jun 3, 2022 at 8:41 PM harshit.varshney@iktara.ai <
> Harshit.Varshney@iktara.ai> wrote:
>
>
>
>
>
> *From:* Shuiqiang Chen [mailto:acqua.csq@gmail.com]
> *Sent:* Friday, June 3, 2022 2:11 PM
> *To:* harshit.varshney@iktara.ai
> *Cc:* user
> *Subject:* Re: Issue Facing While Using EmbeddedRocksDbCheckpointing
> FlinkVersion(1.15.0)
>
>
>
> Hi,
>
>
>
> I guess that the traceback log you provided might not be the root cause of
> the failure, could you please provide the complete log of the Taskmanager？
>
>
>
> Best,
>
> Shuiqiang
>
>
>
> harshit.varshney@iktara.ai <Ha...@iktara.ai> 于2022年6月2日周四 22:04
> 写道：
>
> Dear Team,
>
>
>
> I am new to pyflink and request for your support in issue I am facing with
> Pyflink. I am using Pyflink version 1.15.0 & using reference code from
> pyflink reference code.
>
> The errors I am getting
>
> Traceback (most recent call last):
>
> File
> "E:\pythonProject16\lib\site-packages\apache_beam\runners\worker\data_plane.py",
> line 470, in input_elements
>
> element = received.get(timeout=1)
>
> File
> "C:\Users\Admin\AppData\Local\Programs\Python\Python38\lib\queue.py", line
> 178, in get
>
> raise Empty
>
> _queue.Empty
>
> RuntimeError: Channel closed prematurely.
>
> My code is:
>
> import json
>
> import os
>
> import time
>
> from datetime import datetime
>
>
>
> from pyflink.common import SimpleStringSchema,
> JsonRowDeserializationSchema, Types, JsonRowSerializationSchema
>
> from pyflink.datastream import StreamExecutionEnvironment, WindowFunction,
> HashMapStateBackend, CheckpointingMode, \
>
> FileSystemCheckpointStorage, KeyedProcessFunction, RuntimeContext,
> EmbeddedRocksDBStateBackend, RocksDBStateBackend, \
>
> ExternalizedCheckpointCleanup
>
> from pyflink.datastream.connectors import FlinkKafkaConsumer,
> FlinkKafkaProducer
>
> from pyflink.datastream.state import ValueStateDescriptor, StateTtlConfig,
> ListStateDescriptor
>
> from sklearn.preprocessing import LabelEncoder
>
> import pickle
>
> import pandas as pd
>
> from pyflink.common import Row
>
>
>
> import argparse
>
> from typing import Iterable
>
>
>
> from pyflink.datastream.connectors import FileSink, OutputFileConfig,
> RollingPolicy
>
>
>
> from pyflink.common import Types, WatermarkStrategy, Time, Encoder
>
> from pyflink.common.watermark_strategy import TimestampAssigner
>
> from pyflink.datastream import StreamExecutionEnvironment,
> ProcessWindowFunction
>
> from pyflink.datastream.window import TumblingEventTimeWindows,
> TimeWindow, TumblingProcessingTimeWindows
>
>
>
>
>
> class MyTimestampAssigner(TimestampAssigner):
>
> def extract_timestamp(self, value, record_timestamp) -> int:
>
> return int(value[0])
>
>
>
>
>
> class CountWindowFunction(WindowFunction[tuple, tuple, str, TimeWindow]):
>
> def apply(self, key: str, window: TimeWindow, inputs: Iterable[tuple]):
>
> # return [(key, result)]
>
> return [(key, len([e for e in inputs]))]
>
>
>
>
>
> class Storage(KeyedProcessFunction):
>
>
>
> def __init__(self):
>
> self.state = None
>
>
>
> def open(self, runtime_context: RuntimeContext):
>
> state_descriptor = ValueStateDescriptor("state", Types.FLOAT())
>
> state_ttl_config = StateTtlConfig \
>
> .new_builder(Time.days(7)) \
>
> .set_update_type(StateTtlConfig.UpdateType.OnReadAndWrite) \
>
> .disable_cleanup_in_background() \
>
> .build()
>
> state_descriptor.enable_time_to_live(state_ttl_config)
>
> self.state = runtime_context.get_state(state_descriptor)
>
>
>
> def process_element(self, value, ctx: 'KeyedProcessFunction.Context'):
>
> # retrieve the current count
>
> current = self.state.value()
>
> if current is None:
>
> current = 0
>
> current = value[1]
>
> self.state.update(current)
>
>
>
> yield current,time.time()
>
>
>
>
>
> def write_to_kafka():
>
> env = StreamExecutionEnvironment.get_execution_environment()
>
> env.set_parallelism(1)
>
> env.enable_checkpointing(10000)
>
> env.get_checkpoint_config().set_min_pause_between_checkpoints(5000)
>
>
> env.set_state_backend(EmbeddedRocksDBStateBackend(enable_incremental_checkpointing=True))
>
>
> env.get_checkpoint_config().set_checkpointing_mode(CheckpointingMode.EXACTLY_ONCE)
>
> env.get_checkpoint_config().enable_unaligned_checkpoints()
>
> env.get_checkpoint_config().set_max_concurrent_checkpoints(1)
>
> env.get_checkpoint_config().set_checkpoint_timeout(50000)
>
>
> env.get_checkpoint_config().enable_externalized_checkpoints(ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION)
>
> check = os.path.join(os.path.abspath(os.path.dirname(__file__)),
>
> 'checkpoint-dir11')
>
>
> env.get_checkpoint_config().set_checkpoint_storage(FileSystemCheckpointStorage("
> file:///{}".format(check)))
>
> kafka_jar = os.path.join(os.path.abspath(os.path.dirname(__file__)),
>
> 'flink-sql-connector-kafka_2.11-1.14.4.jar')
>
> env.add_jars("file:///{}".format(kafka_jar))
>
> rocksdb_jar = os.path.join(os.path.abspath(os.path.dirname(__file__)),
>
> 'flink-statebackend-rocksdb_2.11-1.0.0.jar')
>
> env.add_jars("file:///{}".format(rocksdb_jar))
>
> # deserialization_schema = SimpleStringSchema()
>
> deserialization_schema = JsonRowDeserializationSchema.builder() \
>
> .type_info(type_info=Types.ROW_NAMED(["time_stamp",
>
> "Bill_number", "Store_Code",
> "itemdescription", "Item_code",
>
> "Gross_Price", "Discount",
> "Net_Price",
>
> "purchaseorReturn",
>
> "Membership_No",
> "Billing_Date", "Billing_Time"],
>
> [Types.DOUBLE(),
> Types.STRING(), Types.INT(), Types.STRING(),
>
> Types.STRING(),
> Types.FLOAT(), Types.FLOAT(), Types.FLOAT(),
>
> Types.STRING(),
> Types.STRING(), Types.STRING(),
>
> Types.STRING()])).build()
>
>
>
> kafka_consumer = FlinkKafkaConsumer(
>
> topics='test',
>
> deserialization_schema=deserialization_schema,
>
> properties={'bootstrap.servers': '192.168.1.37:9092', 'group.id':
> 'test_group1'})
>
>
>
> ds = env.add_source(kafka_consumer)
>
> # ds.print()
>
> watermark_strategy = WatermarkStrategy.for_monotonous_timestamps() \
>
> .with_timestamp_assigner(MyTimestampAssigner())
>
> # ds = ds.map(lambda x: eval(x))
>
>
>
> ds1 = ds.assign_timestamps_and_watermarks(watermark_strategy) \
>
> .key_by(lambda x: '/root', key_type=Types.STRING()) \
>
> .window(TumblingProcessingTimeWindows.of(Time.seconds(10))) \
>
> .apply(CountWindowFunction(), Types.TUPLE([Types.STRING(),
> Types.INT()]))
>
> ds1.print()
>
> ds4 = ds1.key_by(lambda value: value[0]) \
>
> .process(Storage())
>
> ds4.print()
>
> # ds1 = ds.map(mlfunc)
>
> # ds1.print()
>
>
>
> env.execute('write_to_kafka')
>
>
>
>
>
> if __name__ == '__main__':
>
> print("start writing data to kafka")
>
> write_to_kafka()
>
>
FW: FW: Issue Facing While Using EmbeddedRocksDbCheckpointing FlinkVersion(1.15.0)
Posted by "harshit.varshney@iktara.ai" <Ha...@iktara.ai>.
From: Dian Fu [mailto:dian0511.fu@gmail.com]
Sent: Monday, June 6, 2022 6:47 AM
To: harshit.varshney@iktara.ai
Cc: user
Subject: Re: FW: Issue Facing While Using EmbeddedRocksDbCheckpointing FlinkVersion(1.15.0)
None of the attachments are logs of the TaskManger. The TaskManger log should be located in the directory `E:\pythonProject16\lib\site-packages\pyflink\log`.
On Fri, Jun 3, 2022 at 8:41 PM harshit.varshney@iktara.ai <ma...@iktara.ai> <Harshit.Varshney@iktara.ai <ma...@iktara.ai> > wrote:
From: Shuiqiang Chen [mailto:acqua.csq@gmail.com <ma...@gmail.com> ]
Sent: Friday, June 3, 2022 2:11 PM
To: harshit.varshney@iktara.ai <ma...@iktara.ai>
Cc: user
Subject: Re: Issue Facing While Using EmbeddedRocksDbCheckpointing FlinkVersion(1.15.0)
Hi,
I guess that the traceback log you provided might not be the root cause of the failure, could you please provide the complete log of the Taskmanager？
Best,
Shuiqiang
harshit.varshney@iktara.ai <ma...@iktara.ai> <Harshit.Varshney@iktara.ai <ma...@iktara.ai> > 于2022年6月2日周四 22:04写道：
Dear Team,
I am new to pyflink and request for your support in issue I am facing with Pyflink. I am using Pyflink version 1.15.0 & using reference code from pyflink reference code.
The errors I am getting
Traceback (most recent call last):
File "E:\pythonProject16\lib\site-packages\apache_beam\runners\worker\data_plane.py", line 470, in input_elements
element = received.get(timeout=1)
File "C:\Users\Admin\AppData\Local\Programs\Python\Python38\lib\queue.py", line 178, in get
raise Empty
_queue.Empty
RuntimeError: Channel closed prematurely.
My code is:
import json
import os
import time
from datetime import datetime
from pyflink.common import SimpleStringSchema, JsonRowDeserializationSchema, Types, JsonRowSerializationSchema
from pyflink.datastream import StreamExecutionEnvironment, WindowFunction, HashMapStateBackend, CheckpointingMode, \
FileSystemCheckpointStorage, KeyedProcessFunction, RuntimeContext, EmbeddedRocksDBStateBackend, RocksDBStateBackend, \
ExternalizedCheckpointCleanup
from pyflink.datastream.connectors import FlinkKafkaConsumer, FlinkKafkaProducer
from pyflink.datastream.state import ValueStateDescriptor, StateTtlConfig, ListStateDescriptor
from sklearn.preprocessing import LabelEncoder
import pickle
import pandas as pd
from pyflink.common import Row
import argparse
from typing import Iterable
from pyflink.datastream.connectors import FileSink, OutputFileConfig, RollingPolicy
from pyflink.common import Types, WatermarkStrategy, Time, Encoder
from pyflink.common.watermark_strategy import TimestampAssigner
from pyflink.datastream import StreamExecutionEnvironment, ProcessWindowFunction
from pyflink.datastream.window import TumblingEventTimeWindows, TimeWindow, TumblingProcessingTimeWindows
class MyTimestampAssigner(TimestampAssigner):
def extract_timestamp(self, value, record_timestamp) -> int:
return int(value[0])
class CountWindowFunction(WindowFunction[tuple, tuple, str, TimeWindow]):
def apply(self, key: str, window: TimeWindow, inputs: Iterable[tuple]):
# return [(key, result)]
return [(key, len([e for e in inputs]))]
class Storage(KeyedProcessFunction):
def __init__(self):
self.state = None
def open(self, runtime_context: RuntimeContext):
state_descriptor = ValueStateDescriptor("state", Types.FLOAT())
state_ttl_config = StateTtlConfig \
.new_builder(Time.days(7)) \
.set_update_type(StateTtlConfig.UpdateType.OnReadAndWrite) \
.disable_cleanup_in_background() \
.build()
state_descriptor.enable_time_to_live(state_ttl_config)
self.state = runtime_context.get_state(state_descriptor)
def process_element(self, value, ctx: 'KeyedProcessFunction.Context'):
# retrieve the current count
current = self.state.value()
if current is None:
current = 0
current = value[1]
self.state.update(current)
yield current,time.time()
def write_to_kafka():
env = StreamExecutionEnvironment.get_execution_environment()
env.set_parallelism(1)
env.enable_checkpointing(10000)
env.get_checkpoint_config().set_min_pause_between_checkpoints(5000)
env.set_state_backend(EmbeddedRocksDBStateBackend(enable_incremental_checkpointing=True))
env.get_checkpoint_config().set_checkpointing_mode(CheckpointingMode.EXACTLY_ONCE)
env.get_checkpoint_config().enable_unaligned_checkpoints()
env.get_checkpoint_config().set_max_concurrent_checkpoints(1)
env.get_checkpoint_config().set_checkpoint_timeout(50000)
env.get_checkpoint_config().enable_externalized_checkpoints(ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION)
check = os.path.join(os.path.abspath(os.path.dirname(__file__)),
'checkpoint-dir11')
env.get_checkpoint_config().set_checkpoint_storage(FileSystemCheckpointStorage("file:///{} <file:///\\%7b%7d> ".format(check)))
kafka_jar = os.path.join(os.path.abspath(os.path.dirname(__file__)),
'flink-sql-connector-kafka_2.11-1.14.4.jar')
env.add_jars("file:///{} <file:///\\%7b%7d> ".format(kafka_jar))
rocksdb_jar = os.path.join(os.path.abspath(os.path.dirname(__file__)),
'flink-statebackend-rocksdb_2.11-1.0.0.jar')
env.add_jars("file:///{} <file:///\\%7b%7d> ".format(rocksdb_jar))
# deserialization_schema = SimpleStringSchema()
deserialization_schema = JsonRowDeserializationSchema.builder() \
.type_info(type_info=Types.ROW_NAMED(["time_stamp",
"Bill_number", "Store_Code", "itemdescription", "Item_code",
"Gross_Price", "Discount", "Net_Price",
"purchaseorReturn",
"Membership_No", "Billing_Date", "Billing_Time"],
[Types.DOUBLE(), Types.STRING(), Types.INT(), Types.STRING(),
Types.STRING(), Types.FLOAT(), Types.FLOAT(), Types.FLOAT(),
Types.STRING(), Types.STRING(), Types.STRING(),
Types.STRING()])).build()
kafka_consumer = FlinkKafkaConsumer(
topics='test',
deserialization_schema=deserialization_schema,
properties={'bootstrap.servers': '192.168.1.37:9092 <http://192.168.1.37:9092> ', 'group.id <http://group.id> ': 'test_group1'})
ds = env.add_source(kafka_consumer)
# ds.print()
watermark_strategy = WatermarkStrategy.for_monotonous_timestamps() \
.with_timestamp_assigner(MyTimestampAssigner())
# ds = ds.map(lambda x: eval(x))
ds1 = ds.assign_timestamps_and_watermarks(watermark_strategy) \
.key_by(lambda x: '/root', key_type=Types.STRING()) \
.window(TumblingProcessingTimeWindows.of(Time.seconds(10))) \
.apply(CountWindowFunction(), Types.TUPLE([Types.STRING(), Types.INT()]))
ds1.print()
ds4 = ds1.key_by(lambda value: value[0]) \
.process(Storage())
ds4.print()
# ds1 = ds.map(mlfunc)
# ds1.print()
env.execute('write_to_kafka')
if __name__ == '__main__':
print("start writing data to kafka")
write_to_kafka()
Re: FW: Issue Facing While Using EmbeddedRocksDbCheckpointing FlinkVersion(1.15.0)
Posted by Dian Fu <di...@gmail.com>.
None of the attachments are logs of the TaskManger. The TaskManger log
should be located in the directory
`E:\pythonProject16\lib\site-packages\pyflink\log`.
On Fri, Jun 3, 2022 at 8:41 PM harshit.varshney@iktara.ai <
Harshit.Varshney@iktara.ai> wrote:
>
>
>
>
> *From:* Shuiqiang Chen [mailto:acqua.csq@gmail.com]
> *Sent:* Friday, June 3, 2022 2:11 PM
> *To:* harshit.varshney@iktara.ai
> *Cc:* user
> *Subject:* Re: Issue Facing While Using EmbeddedRocksDbCheckpointing
> FlinkVersion(1.15.0)
>
>
>
> Hi,
>
>
>
> I guess that the traceback log you provided might not be the root cause of
> the failure, could you please provide the complete log of the Taskmanager？
>
>
>
> Best,
>
> Shuiqiang
>
>
>
> harshit.varshney@iktara.ai <Ha...@iktara.ai> 于2022年6月2日周四 22:04
> 写道：
>
> Dear Team,
>
>
>
> I am new to pyflink and request for your support in issue I am facing with
> Pyflink. I am using Pyflink version 1.15.0 & using reference code from
> pyflink reference code.
>
> The errors I am getting
>
> Traceback (most recent call last):
>
> File
> "E:\pythonProject16\lib\site-packages\apache_beam\runners\worker\data_plane.py",
> line 470, in input_elements
>
> element = received.get(timeout=1)
>
> File
> "C:\Users\Admin\AppData\Local\Programs\Python\Python38\lib\queue.py", line
> 178, in get
>
> raise Empty
>
> _queue.Empty
>
> RuntimeError: Channel closed prematurely.
>
> My code is:
>
> import json
>
> import os
>
> import time
>
> from datetime import datetime
>
>
>
> from pyflink.common import SimpleStringSchema,
> JsonRowDeserializationSchema, Types, JsonRowSerializationSchema
>
> from pyflink.datastream import StreamExecutionEnvironment, WindowFunction,
> HashMapStateBackend, CheckpointingMode, \
>
> FileSystemCheckpointStorage, KeyedProcessFunction, RuntimeContext,
> EmbeddedRocksDBStateBackend, RocksDBStateBackend, \
>
> ExternalizedCheckpointCleanup
>
> from pyflink.datastream.connectors import FlinkKafkaConsumer,
> FlinkKafkaProducer
>
> from pyflink.datastream.state import ValueStateDescriptor, StateTtlConfig,
> ListStateDescriptor
>
> from sklearn.preprocessing import LabelEncoder
>
> import pickle
>
> import pandas as pd
>
> from pyflink.common import Row
>
>
>
> import argparse
>
> from typing import Iterable
>
>
>
> from pyflink.datastream.connectors import FileSink, OutputFileConfig,
> RollingPolicy
>
>
>
> from pyflink.common import Types, WatermarkStrategy, Time, Encoder
>
> from pyflink.common.watermark_strategy import TimestampAssigner
>
> from pyflink.datastream import StreamExecutionEnvironment,
> ProcessWindowFunction
>
> from pyflink.datastream.window import TumblingEventTimeWindows,
> TimeWindow, TumblingProcessingTimeWindows
>
>
>
>
>
> class MyTimestampAssigner(TimestampAssigner):
>
> def extract_timestamp(self, value, record_timestamp) -> int:
>
> return int(value[0])
>
>
>
>
>
> class CountWindowFunction(WindowFunction[tuple, tuple, str, TimeWindow]):
>
> def apply(self, key: str, window: TimeWindow, inputs: Iterable[tuple]):
>
> # return [(key, result)]
>
> return [(key, len([e for e in inputs]))]
>
>
>
>
>
> class Storage(KeyedProcessFunction):
>
>
>
> def __init__(self):
>
> self.state = None
>
>
>
> def open(self, runtime_context: RuntimeContext):
>
> state_descriptor = ValueStateDescriptor("state", Types.FLOAT())
>
> state_ttl_config = StateTtlConfig \
>
> .new_builder(Time.days(7)) \
>
> .set_update_type(StateTtlConfig.UpdateType.OnReadAndWrite) \
>
> .disable_cleanup_in_background() \
>
> .build()
>
> state_descriptor.enable_time_to_live(state_ttl_config)
>
> self.state = runtime_context.get_state(state_descriptor)
>
>
>
> def process_element(self, value, ctx: 'KeyedProcessFunction.Context'):
>
> # retrieve the current count
>
> current = self.state.value()
>
> if current is None:
>
> current = 0
>
> current = value[1]
>
> self.state.update(current)
>
>
>
> yield current,time.time()
>
>
>
>
>
> def write_to_kafka():
>
> env = StreamExecutionEnvironment.get_execution_environment()
>
> env.set_parallelism(1)
>
> env.enable_checkpointing(10000)
>
> env.get_checkpoint_config().set_min_pause_between_checkpoints(5000)
>
>
> env.set_state_backend(EmbeddedRocksDBStateBackend(enable_incremental_checkpointing=True))
>
>
> env.get_checkpoint_config().set_checkpointing_mode(CheckpointingMode.EXACTLY_ONCE)
>
> env.get_checkpoint_config().enable_unaligned_checkpoints()
>
> env.get_checkpoint_config().set_max_concurrent_checkpoints(1)
>
> env.get_checkpoint_config().set_checkpoint_timeout(50000)
>
>
> env.get_checkpoint_config().enable_externalized_checkpoints(ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION)
>
> check = os.path.join(os.path.abspath(os.path.dirname(__file__)),
>
> 'checkpoint-dir11')
>
>
> env.get_checkpoint_config().set_checkpoint_storage(FileSystemCheckpointStorage("file:///{}".format(check)))
>
> kafka_jar = os.path.join(os.path.abspath(os.path.dirname(__file__)),
>
> 'flink-sql-connector-kafka_2.11-1.14.4.jar')
>
> env.add_jars("file:///{}".format(kafka_jar))
>
> rocksdb_jar = os.path.join(os.path.abspath(os.path.dirname(__file__)),
>
> 'flink-statebackend-rocksdb_2.11-1.0.0.jar')
>
> env.add_jars("file:///{}".format(rocksdb_jar))
>
> # deserialization_schema = SimpleStringSchema()
>
> deserialization_schema = JsonRowDeserializationSchema.builder() \
>
> .type_info(type_info=Types.ROW_NAMED(["time_stamp",
>
> "Bill_number", "Store_Code",
> "itemdescription", "Item_code",
>
> "Gross_Price", "Discount",
> "Net_Price",
>
> "purchaseorReturn",
>
> "Membership_No",
> "Billing_Date", "Billing_Time"],
>
> [Types.DOUBLE(),
> Types.STRING(), Types.INT(), Types.STRING(),
>
> Types.STRING(),
> Types.FLOAT(), Types.FLOAT(), Types.FLOAT(),
>
> Types.STRING(),
> Types.STRING(), Types.STRING(),
>
> Types.STRING()])).build()
>
>
>
> kafka_consumer = FlinkKafkaConsumer(
>
> topics='test',
>
> deserialization_schema=deserialization_schema,
>
> properties={'bootstrap.servers': '192.168.1.37:9092', 'group.id':
> 'test_group1'})
>
>
>
> ds = env.add_source(kafka_consumer)
>
> # ds.print()
>
> watermark_strategy = WatermarkStrategy.for_monotonous_timestamps() \
>
> .with_timestamp_assigner(MyTimestampAssigner())
>
> # ds = ds.map(lambda x: eval(x))
>
>
>
> ds1 = ds.assign_timestamps_and_watermarks(watermark_strategy) \
>
> .key_by(lambda x: '/root', key_type=Types.STRING()) \
>
> .window(TumblingProcessingTimeWindows.of(Time.seconds(10))) \
>
> .apply(CountWindowFunction(), Types.TUPLE([Types.STRING(),
> Types.INT()]))
>
> ds1.print()
>
> ds4 = ds1.key_by(lambda value: value[0]) \
>
> .process(Storage())
>
> ds4.print()
>
> # ds1 = ds.map(mlfunc)
>
> # ds1.print()
>
>
>
> env.execute('write_to_kafka')
>
>
>
>
>
> if __name__ == '__main__':
>
> print("start writing data to kafka")
>
> write_to_kafka()
>
>
FW: Issue Facing While Using EmbeddedRocksDbCheckpointing FlinkVersion(1.15.0)
Posted by "harshit.varshney@iktara.ai" <Ha...@iktara.ai>.
From: Shuiqiang Chen [mailto:acqua.csq@gmail.com]
Sent: Friday, June 3, 2022 2:11 PM
To: harshit.varshney@iktara.ai
Cc: user
Subject: Re: Issue Facing While Using EmbeddedRocksDbCheckpointing FlinkVersion(1.15.0)
Hi,
I guess that the traceback log you provided might not be the root cause of the failure, could you please provide the complete log of the Taskmanager？
Best,
Shuiqiang
harshit.varshney@iktara.ai <ma...@iktara.ai> <Harshit.Varshney@iktara.ai <ma...@iktara.ai> > 于2022年6月2日周四 22:04写道：
Dear Team,
I am new to pyflink and request for your support in issue I am facing with Pyflink. I am using Pyflink version 1.15.0 & using reference code from pyflink reference code.
The errors I am getting
Traceback (most recent call last):
File "E:\pythonProject16\lib\site-packages\apache_beam\runners\worker\data_plane.py", line 470, in input_elements
element = received.get(timeout=1)
File "C:\Users\Admin\AppData\Local\Programs\Python\Python38\lib\queue.py", line 178, in get
raise Empty
_queue.Empty
RuntimeError: Channel closed prematurely.
My code is:
import json
import os
import time
from datetime import datetime
from pyflink.common import SimpleStringSchema, JsonRowDeserializationSchema, Types, JsonRowSerializationSchema
from pyflink.datastream import StreamExecutionEnvironment, WindowFunction, HashMapStateBackend, CheckpointingMode, \
FileSystemCheckpointStorage, KeyedProcessFunction, RuntimeContext, EmbeddedRocksDBStateBackend, RocksDBStateBackend, \
ExternalizedCheckpointCleanup
from pyflink.datastream.connectors import FlinkKafkaConsumer, FlinkKafkaProducer
from pyflink.datastream.state import ValueStateDescriptor, StateTtlConfig, ListStateDescriptor
from sklearn.preprocessing import LabelEncoder
import pickle
import pandas as pd
from pyflink.common import Row
import argparse
from typing import Iterable
from pyflink.datastream.connectors import FileSink, OutputFileConfig, RollingPolicy
from pyflink.common import Types, WatermarkStrategy, Time, Encoder
from pyflink.common.watermark_strategy import TimestampAssigner
from pyflink.datastream import StreamExecutionEnvironment, ProcessWindowFunction
from pyflink.datastream.window import TumblingEventTimeWindows, TimeWindow, TumblingProcessingTimeWindows
class MyTimestampAssigner(TimestampAssigner):
def extract_timestamp(self, value, record_timestamp) -> int:
return int(value[0])
class CountWindowFunction(WindowFunction[tuple, tuple, str, TimeWindow]):
def apply(self, key: str, window: TimeWindow, inputs: Iterable[tuple]):
# return [(key, result)]
return [(key, len([e for e in inputs]))]
class Storage(KeyedProcessFunction):
def __init__(self):
self.state = None
def open(self, runtime_context: RuntimeContext):
state_descriptor = ValueStateDescriptor("state", Types.FLOAT())
state_ttl_config = StateTtlConfig \
.new_builder(Time.days(7)) \
.set_update_type(StateTtlConfig.UpdateType.OnReadAndWrite) \
.disable_cleanup_in_background() \
.build()
state_descriptor.enable_time_to_live(state_ttl_config)
self.state = runtime_context.get_state(state_descriptor)
def process_element(self, value, ctx: 'KeyedProcessFunction.Context'):
# retrieve the current count
current = self.state.value()
if current is None:
current = 0
current = value[1]
self.state.update(current)
yield current,time.time()
def write_to_kafka():
env = StreamExecutionEnvironment.get_execution_environment()
env.set_parallelism(1)
env.enable_checkpointing(10000)
env.get_checkpoint_config().set_min_pause_between_checkpoints(5000)
env.set_state_backend(EmbeddedRocksDBStateBackend(enable_incremental_checkpointing=True))
env.get_checkpoint_config().set_checkpointing_mode(CheckpointingMode.EXACTLY_ONCE)
env.get_checkpoint_config().enable_unaligned_checkpoints()
env.get_checkpoint_config().set_max_concurrent_checkpoints(1)
env.get_checkpoint_config().set_checkpoint_timeout(50000)
env.get_checkpoint_config().enable_externalized_checkpoints(ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION)
check = os.path.join(os.path.abspath(os.path.dirname(__file__)),
'checkpoint-dir11')
env.get_checkpoint_config().set_checkpoint_storage(FileSystemCheckpointStorage("file:///{}".format(check)))
kafka_jar = os.path.join(os.path.abspath(os.path.dirname(__file__)),
'flink-sql-connector-kafka_2.11-1.14.4.jar')
env.add_jars("file:///{}".format(kafka_jar))
rocksdb_jar = os.path.join(os.path.abspath(os.path.dirname(__file__)),
'flink-statebackend-rocksdb_2.11-1.0.0.jar')
env.add_jars("file:///{}".format(rocksdb_jar))
# deserialization_schema = SimpleStringSchema()
deserialization_schema = JsonRowDeserializationSchema.builder() \
.type_info(type_info=Types.ROW_NAMED(["time_stamp",
"Bill_number", "Store_Code", "itemdescription", "Item_code",
"Gross_Price", "Discount", "Net_Price",
"purchaseorReturn",
"Membership_No", "Billing_Date", "Billing_Time"],
[Types.DOUBLE(), Types.STRING(), Types.INT(), Types.STRING(),
Types.STRING(), Types.FLOAT(), Types.FLOAT(), Types.FLOAT(),
Types.STRING(), Types.STRING(), Types.STRING(),
Types.STRING()])).build()
kafka_consumer = FlinkKafkaConsumer(
topics='test',
deserialization_schema=deserialization_schema,
properties={'bootstrap.servers': '192.168.1.37:9092', 'group.id': 'test_group1'})
ds = env.add_source(kafka_consumer)
# ds.print()
watermark_strategy = WatermarkStrategy.for_monotonous_timestamps() \
.with_timestamp_assigner(MyTimestampAssigner())
# ds = ds.map(lambda x: eval(x))
ds1 = ds.assign_timestamps_and_watermarks(watermark_strategy) \
.key_by(lambda x: '/root', key_type=Types.STRING()) \
.window(TumblingProcessingTimeWindows.of(Time.seconds(10))) \
.apply(CountWindowFunction(), Types.TUPLE([Types.STRING(), Types.INT()]))
ds1.print()
ds4 = ds1.key_by(lambda value: value[0]) \
.process(Storage())
ds4.print()
# ds1 = ds.map(mlfunc)
# ds1.print()
env.execute('write_to_kafka')
if __name__ == '__main__':
print("start writing data to kafka")
write_to_kafka()
Re: Issue Facing While Using EmbeddedRocksDbCheckpointing FlinkVersion(1.15.0)
Posted by Shuiqiang Chen <ac...@gmail.com>.
Hi,
I guess that the traceback log you provided might not be the root cause of
the failure, could you please provide the complete log of the Taskmanager？
Best,
Shuiqiang
harshit.varshney@iktara.ai <Ha...@iktara.ai> 于2022年6月2日周四
22:04写道：
> Dear Team,
>
>
>
> I am new to pyflink and request for your support in issue I am facing with
> Pyflink. I am using Pyflink version 1.15.0 & using reference code from
> pyflink reference code.
>
> The errors I am getting
>
> Traceback (most recent call last):
>
> File
> "E:\pythonProject16\lib\site-packages\apache_beam\runners\worker\data_plane.py",
> line 470, in input_elements
>
> element = received.get(timeout=1)
>
> File
> "C:\Users\Admin\AppData\Local\Programs\Python\Python38\lib\queue.py", line
> 178, in get
>
> raise Empty
>
> _queue.Empty
>
> RuntimeError: Channel closed prematurely.
>
> My code is:
>
> import json
>
> import os
>
> import time
>
> from datetime import datetime
>
>
>
> from pyflink.common import SimpleStringSchema,
> JsonRowDeserializationSchema, Types, JsonRowSerializationSchema
>
> from pyflink.datastream import StreamExecutionEnvironment, WindowFunction,
> HashMapStateBackend, CheckpointingMode, \
>
> FileSystemCheckpointStorage, KeyedProcessFunction, RuntimeContext,
> EmbeddedRocksDBStateBackend, RocksDBStateBackend
>
> from pyflink.datastream.connectors import FlinkKafkaConsumer,
> FlinkKafkaProducer
>
> from pyflink.datastream.state import ValueStateDescriptor, StateTtlConfig,
> ListStateDescriptor
>
> from sklearn.preprocessing import LabelEncoder
>
> import pickle
>
> import pandas as pd
>
> from pyflink.common import Row
>
>
>
> import argparse
>
> from typing import Iterable
>
>
>
> from pyflink.datastream.connectors import FileSink, OutputFileConfig,
> RollingPolicy
>
>
>
> from pyflink.common import Types, WatermarkStrategy, Time, Encoder
>
> from pyflink.common.watermark_strategy import TimestampAssigner
>
> from pyflink.datastream import StreamExecutionEnvironment,
> ProcessWindowFunction
>
> from pyflink.datastream.window import TumblingEventTimeWindows,
> TimeWindow, TumblingProcessingTimeWindows
>
>
>
>
>
> class MyTimestampAssigner(TimestampAssigner):
>
> def extract_timestamp(self, value, record_timestamp) -> int:
>
> return int(value[0])
>
>
>
>
>
> class CountWindowFunction(WindowFunction[tuple, tuple, str, TimeWindow]):
>
> def apply(self, key: str, window: TimeWindow, inputs: Iterable[tuple]):
>
> # return [(key, result)]
>
> return [(key, len([e for e in inputs]))]
>
>
>
>
>
> class Storage(KeyedProcessFunction):
>
>
>
> def __init__(self):
>
> self.state = None
>
>
>
> def open(self, runtime_context: RuntimeContext):
>
> state_descriptor = ValueStateDescriptor("state", Types.FLOAT())
>
> state_ttl_config = StateTtlConfig \
>
> .new_builder(Time.days(7)) \
>
> .set_update_type(StateTtlConfig.UpdateType.OnReadAndWrite) \
>
> .disable_cleanup_in_background() \
>
> .build()
>
> state_descriptor.enable_time_to_live(state_ttl_config)
>
> self.state = runtime_context.get_state(state_descriptor)
>
>
>
> def process_element(self, value, ctx: 'KeyedProcessFunction.Context'):
>
> # retrieve the current count
>
> current = self.state.value()
>
> if current is None:
>
> current = 0
>
> current = value[1]
>
> self.state.update(current)
>
>
>
> yield current,time.time()
>
>
>
>
>
> def write_to_kafka():
>
> env = StreamExecutionEnvironment.get_execution_environment()
>
> env.set_parallelism(1)
>
> env.enable_checkpointing(1000)
>
> env.get_checkpoint_config().set_min_pause_between_checkpoints(5000)
>
> env.set_state_backend(EmbeddedRocksDBStateBackend())
>
>
> env.get_checkpoint_config().set_checkpointing_mode(CheckpointingMode.AT_LEAST_ONCE)
>
> #env.get_checkpoint_config().enable_unaligned_checkpoints()
>
> check = os.path.join(os.path.abspath(os.path.dirname(__file__)),
>
> 'checkpoint-dir11')
>
>
> env.get_checkpoint_config().set_checkpoint_storage(FileSystemCheckpointStorage("file:///{}".format(check)))
>
> kafka_jar = os.path.join(os.path.abspath(os.path.dirname(__file__)),
>
> 'flink-sql-connector-kafka_2.11-1.14.4.jar')
>
> env.add_jars("file:///{}".format(kafka_jar))
>
> rocksdb_jar = os.path.join(os.path.abspath(os.path.dirname(__file__)),
>
> 'flink-statebackend-rocksdb_2.11-1.0.0.jar')
>
> env.add_jars("file:///{}".format(rocksdb_jar))
>
> # deserialization_schema = SimpleStringSchema()
>
> deserialization_schema = JsonRowDeserializationSchema.builder() \
>
> .type_info(type_info=Types.ROW_NAMED(["time_stamp",
>
> "Bill_number", "Store_Code",
> "itemdescription", "Item_code",
>
> "Gross_Price", "Discount",
> "Net_Price",
>
> "purchaseorReturn",
>
> "Membership_No",
> "Billing_Date", "Billing_Time"],
>
> [Types.DOUBLE(),
> Types.STRING(), Types.INT(), Types.STRING(),
>
> Types.STRING(),
> Types.FLOAT(), Types.FLOAT(), Types.FLOAT(),
>
> Types.STRING(),
> Types.STRING(), Types.STRING(),
>
> Types.STRING()])).build()
>
>
>
> kafka_consumer = FlinkKafkaConsumer(
>
> topics='test',
>
> deserialization_schema=deserialization_schema,
>
> properties={'bootstrap.servers': '192.168.1.37:9092', 'group.id':
> 'test_group1'})
>
>
>
> ds = env.add_source(kafka_consumer)
>
> # ds.print()
>
> watermark_strategy = WatermarkStrategy.for_monotonous_timestamps() \
>
> .with_timestamp_assigner(MyTimestampAssigner())
>
> # ds = ds.map(lambda x: eval(x))
>
>
>
> ds1 = ds.assign_timestamps_and_watermarks(watermark_strategy) \
>
> .key_by(lambda x: '/root', key_type=Types.STRING()) \
>
> .window(TumblingProcessingTimeWindows.of(Time.seconds(1))) \
>
> .apply(CountWindowFunction(), Types.TUPLE([Types.STRING(),
> Types.INT()]))
>
> ds1.print()
>
> ds4 = ds1.key_by(lambda value: value[0]) \
>
> .process(Storage())
>
> ds4.print()
>
> # ds1 = ds.map(mlfunc)
>
> # ds1.print()
>
>
>
> env.execute('write_to_kafka')
>
>
>
>
>
> if __name__ == '__main__':
>
> print("start writing data to kafka")
>
> write_to_kafka()
>
>
>
>
>