我试图向运行在kubernetes集群上的flink作业管理器发送post请求。当为不需要任何命令行参数的类发送/jar/run的post请求时,它可以正常工作。但是,当尝试在同一个jar中提交需要命令行参数的不同类时,会出现以下错误。-: {"errors":["Request did not match expected format JarRunRequestBody."]}'
但是,在传递命令行参数和直接提交作业时,请执行以下操作-:
./flink run -m localhost:30287 -c com.class.name ~/path/to/jar/1.0-1.0-SNAPSHOT.jar --bootstrap.servers izac-cp-kafka:9092 --group.id test --topic bank_transaction --schema.registry http://mysr-schema-registry:8081 --CepJson """{\"keyId\": \"customer_id\",\"pattern\": [{\"patternName\": \"p1\",\"simpleCondition\":{\"columnName\": \"amount\",\"operator\": \">\",\"value\": \"50\",\"dataType\": \"Int\"}},{\"patternName\":\"p2\",\"simpleCondition\":{\"columnName\":\"amount\",\"operator\":\">\",\"value\":\"30\",\"dataType\":\"Int\"}}],\"connector\":[{\"name\":\"begin\",\"connectorType\":\"next\",\"start\":\"p1\",\"end\":\"p2\"}]}"""
要将上述命令转换为基于flink rest的post请求,我执行了以下操作-:
curl -k -v -X POST -H "Content-Type: application/json" --data '{ "entryClass":"com.class.name", "programArgsList": [ "--bootstrap.servers izac-cp-kafka:9092", "--group.id test", "--topic bank_transaction", "--schema.registry http://mysr-schema-registry:8081", "--CepJson """{\"keyId\": \"customer_id\",\"pattern\": [{\"patternName\": \"p1\",\"simpleCondition\":{\"columnName\": \"amount\",\"operator\": \">\",\"value\": \"50\",\"dataType\": \"Int\"}},{\"patternName\":\"p2\",\"simpleCondition\":{\"columnName\":\"amount\",\"operator\":\">\",\"value\":\"30\",\"dataType\":\"Int\"}}],\"connector\":[{\"name\":\"begin\",\"connectorType\":\"next\",\"start\":\"p1\",\"end\":\"p2\"}]}""""]}' http://localhost:30287/jars/2a788e33-c92d-47c4-84af-31e3dff28666_1.0-1.0-SNAPSHOT.jar/run
但是,这就产生了前面提到的错误。我只是想将上面的命令行作业提交转换为基于restapi的对flink集群的提交。
注意-:post请求是针对已经包含所需jar的flink集群的。我只想提交使用特定类的作业。
2条答案
按热度按时间vmpqdwk31#
我将上面的curl请求更改为以下内容,它成功了-:
m3eecexj2#
我不认为卷发接受
"""
作为一个字符串插值的方式,scala这样做,它不会发送正确的CepJson
参数是肯定的,所以我要从改变它开始。