Ufdjrw

For a first try I want to read JSON data from a file and pass it on to Flink. I defined a source (which reads JSON strings line by line) and a placeholder filter. See Code:

from org.apache.flink.streaming.api.functions.source import SourceFunction
from org.apache.flink.api.common.functions import FilterFunction
import json
import sys

class Json_reader(SourceFunction):
 def readjason(self, ctx):
 sys.stdin = open('capture.json', 'r')
 for line in sys.stdin:
 ctx.collect(json.loads(line))


class Dummy_Filter(FilterFunction):
 def filter(self, value):
 return True

#
# The pipeline definition.
#
def main(factory):
 env = factory.get_execution_environment()
 env.create_python_source(Json_reader()) 
 .filter(Dummy_Filter()) 
 .output()
 env.execute()

When I build the job and move it to my started Flink-cluster, I get the following error message:

VirtualBox:/media/sf_Python$ ./flink-1.7.2/bin/pyflink-stream.sh
./json_parser_flink.py Starting execution of program Failed to run
plan: null Traceback (most recent call last): File "", line
1, in File
"/tmp/flink_streaming_plan_fbe13c4c-6918-46d4-a4bc-36908a2bea24/json_parser_flink.py",
line 25, in main at
org.apache.flink.client.program.rest.RestClusterClient.submitJob(RestClusterClient.java:268)
at
org.apache.flink.client.program.ClusterClient.run(ClusterClient.java:487)
at
org.apache.flink.streaming.api.environment.StreamContextEnvironment.execute(StreamContextEnvironment.java:66)
at
org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.execute(StreamExecutionEnvironment.java:1510)
at
org.apache.flink.streaming.python.api.environment.PythonStreamExecutionEnvironment.execute(PythonStreamExecutionEnvironment.java:245)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) at
sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at
sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
org.apache.flink.client.program.ProgramInvocationException:
org.apache.flink.client.program.ProgramInvocationException: Job
failed. (JobID: 31615948194c951be03d46576929aa23)

The program didn't contain a Flink job. Perhaps you forgot to call
execute() on the execution environment.

I haven't forgotten to call execute().

I found the problem. Fast expects a run() function in the SourceFunction.