Hello everybody. I am new in <https://nightlies.ap...
# troubleshootinga
Amir Hossein Sharifzadeh
02/07/2023, 11:08 PMHello everybody. I am new in (Py)Flink. In order to understand it better, I was trying to run an existing project from: https://apache.googlesource.com/flink-playgrounds/. I could run it on my docker successfully. But still, I have questions: 1) First: In payment_msg_proccessing.py code, I want to run a simple query on Kafka stream (payment_msg table) without insertion data into the sink table (es_sink here), and do some data processing. (In my project, I won’t insert any data). So, is it possible to run the query (queries) on sources (streams) without insertion data into other tables? 2) Second: How can I iterate over results, and print data in the output? For example, I wrote this simple query: table_result = t_env.execute_sql(“select provinceId, payAmount from payment_msg”) then after:
Copy code
with table_result.collect() as results:
for result in results:
print(result)d
Dian Fu
02/08/2023, 5:34 AMRe 1) Could you describe a bit why there is no sink in your job? Are you running some batch queries?
Re 2) The code seems correct for me. Could you explain more about why you think it doesn’t work? Have you seen any exceptions or there is no output and so you think it doesn’t work.
a
Amir Hossein Sharifzadeh
02/08/2023, 4:12 PMHi Dian, As I said I am running a code from the GitBub payment_msg_proccessing.py
Copy code
create_kafka_source_ddl = """
CREATE TABLE payment_msg(
createTime VARCHAR,
orderId BIGINT,
payAmount DOUBLE,
payPlatform INT,
provinceId INT
) WITH (
'connector' = 'kafka',
'topic' = 'payment_msg',
'properties.bootstrap.servers' = 'kafka:9092',
'properties.group.id' = 'test_3',
'scan.startup.mode' = 'latest-offset',
'format' = 'json'
)
""" Copy code
t_env.execute_sql("select * from payment_msg").print()Amir Hossein Sharifzadeh
02/08/2023, 4:12 PMat org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleMessage(AkkaRpcActor.java:168)
at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:24)
at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:20)
at scala.PartialFunction.applyOrElse(PartialFunction.scala:123)
at scala.PartialFunction.applyOrElse$(PartialFunction.scala:122)
at akka.japi.pf.UnitCaseStatement.applyOrElse(CaseStatements.scala:20)
at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:171)
at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:172)
at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:172)
at akka.actor.Actor.aroundReceive(Actor.scala:537)
at akka.actor.Actor.aroundReceive$(Actor.scala:535)
at akka.actor.AbstractActor.aroundReceive(AbstractActor.scala:220)
at akka.actor.ActorCell.receiveMessage(ActorCell.scala:580)
at akka.actor.ActorCell.invoke(ActorCell.scala:548)
at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:270)
at akka.dispatch.Mailbox.run(Mailbox.scala:231)
at akka.dispatch.Mailbox.exec(Mailbox.scala:243)
at java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:289)
at java.util.concurrent.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1056)
at java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1692)
at java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:175)
Caused by: java.util.concurrent.CompletionException: java.util.concurrent.CompletionException: org.apache.flink.runtime.jobmanager.scheduler.NoResourceAvailableException: Could not acquire the minimum required resources.
at org.apache.flink.runtime.scheduler.DefaultExecutionDeployer.lambda$assignResource$4(DefaultExecutionDeployer.java:227)
... 38 more
Caused by: java.util.concurrent.CompletionException: org.apache.flink.runtime.jobmanager.scheduler.NoResourceAvailableException: Could not acquire the minimum required resources.
at java.util.concurrent.CompletableFuture.encodeThrowable(CompletableFuture.java:292)
at java.util.concurrent.CompletableFuture.completeThrowable(CompletableFuture.java:308)
at java.util.concurrent.CompletableFuture.uniApply(CompletableFuture.java:607)
at java.util.concurrent.CompletableFuture$UniApply.tryFire(CompletableFuture.java:591)
... 36 more
Caused by: org.apache.flink.runtime.jobmanager.scheduler.NoResourceAvailableException: Could not acquire the minimum required resources.
org.apache.flink.client.program.ProgramAbortException: java.lang.RuntimeException: Python process exits with code: 1
at org.apache.flink.client.python.PythonDriver.main(PythonDriver.java:140)
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)
at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:355)
at org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:222)
at org.apache.flink.client.ClientUtils.executeProgram(ClientUtils.java:98)
at org.apache.flink.client.cli.CliFrontend.executeProgram(CliFrontend.java:846)
at org.apache.flink.client.cli.CliFrontend.run(CliFrontend.java:240)
at org.apache.flink.client.cli.CliFrontend.parseAndRun(CliFrontend.java:1090)
at org.apache.flink.client.cli.CliFrontend.lambda$main$10(CliFrontend.java:1168)
at org.apache.flink.runtime.security.contexts.NoOpSecurityContext.runSecured(NoOpSecurityContext.java:28)
at org.apache.flink.client.cli.CliFrontend.main(CliFrontend.java:1168)
Caused by: java.lang.RuntimeException: Python process exits with code: 1
at org.apache.flink.client.python.PythonDriver.main(PythonDriver.java:130)
... 13 more
Amir Hossein Sharifzadeh
02/08/2023, 4:13 PMI run this command:
Amir Hossein Sharifzadeh
02/08/2023, 4:14 PM Copy code
docker - compose exec jobmanager. / bin / flink run - py / opt / pyflink - walkthrough / payment_msg_proccessing.py - dd
Dian Fu
02/09/2023, 2:13 AMOK. Then this is same issue you reported in another thread. There is no enough resource in the Flink cluster. You can check if there are any TaskManagers running in the Flink cluster before submitting the job via the Flink web ui: http://localhost:8081. If not, you can restart the Flink cluster.