防制网站怎么做,做网站一般需要多少钱,设置自动删除的wordpress,活动推广软文文章目录 背景简单例子flink 例子 背景
代码生成技术适合在需要动态代码编译的场景中使用。比如大数据计算场景下#xff0c;经常会要把flink sql 转成实际的执行计划
简单例子
代码是一个string 类型#xff0c;直接用janino 编译后#xff0c;就可以得到加载到jvm里的c… 文章目录 背景简单例子flink 例子 背景
代码生成技术适合在需要动态代码编译的场景中使用。比如大数据计算场景下经常会要把flink sql 转成实际的执行计划
简单例子
代码是一个string 类型直接用janino 编译后就可以得到加载到jvm里的class 类
import org.codehaus.janino.SimpleCompiler;public class JaninoClassExample {public static void main(String[] args) throws Exception {// 创建一个 SimpleCompilerSimpleCompiler compiler new SimpleCompiler();// 设置 Java 类代码String classCode public class HelloWorld { public String greet(String name) { return \Hello, \ name \!\; } };// 编译 Java 类代码compiler.cook(classCode);// 获取编译后的类Class? helloWorldClass compiler.getClassLoader().loadClass(HelloWorld);// 创建类的实例并调用方法Object helloWorldInstance helloWorldClass.getDeclaredConstructor().newInstance();String greeting (String) helloWorldClass.getMethod(greet, String.class).invoke(helloWorldInstance, Janino);System.out.println(greeting); // 输出 Hello, Janino!}
}
flink 例子
无group by 的sum 函数最后成的实现代码这个类AggCodeGenHelper 里
public class LocalNoGroupingAggregateWithoutKeys$6 extends org.apache.flink.table.runtime.operators.TableStreamOperatorimplements org.apache.flink.streaming.api.operators.OneInputStreamOperator, org.apache.flink.streaming.api.operators.BoundedOneInput {private final Object[] references; // 引用对象数组外部依赖int hash_agg0_sum; // 聚合和boolean hash_agg0_sumIsNull; // 标记聚合和是否为nullorg.apache.flink.table.data.GenericRowData valueRow$5 new org.apache.flink.table.data.GenericRowData(1); // 保存输出值的行数据private boolean hasInput false; // 标记是否有输入org.apache.flink.streaming.runtime.streamrecord.StreamRecord element new org.apache.flink.streaming.runtime.streamrecord.StreamRecord((Object) null); // 输入元素private final org.apache.flink.streaming.runtime.streamrecord.StreamRecord outElement new org.apache.flink.streaming.runtime.streamrecord.StreamRecord(null); // 输出元素// 构造函数public LocalNoGroupingAggregateWithoutKeys$6(Object[] references,org.apache.flink.streaming.runtime.tasks.StreamTask task,org.apache.flink.streaming.api.graph.StreamConfig config,org.apache.flink.streaming.api.operators.Output output,org.apache.flink.streaming.runtime.tasks.ProcessingTimeService processingTimeService) throws Exception {this.references references;this.setup(task, config, output); // 设置算子if (this instanceof org.apache.flink.streaming.api.operators.AbstractStreamOperator) {((org.apache.flink.streaming.api.operators.AbstractStreamOperator) this).setProcessingTimeService(processingTimeService); // 设置处理时间服务}}Overridepublic void open() throws Exception {super.open(); // 调用父类的open方法}Overridepublic void processElement(org.apache.flink.streaming.runtime.streamrecord.StreamRecord element) throws Exception {org.apache.flink.table.data.RowData in1 (org.apache.flink.table.data.RowData) element.getValue(); // 获取输入行数据int field$0; // 输入字段的值boolean isNull$0; // 标记字段值是否为nullboolean isNull$1; // 中间null标记int result$2; // 中间结果if (!hasInput) { // 如果没有输入hasInput true;// 初始化聚合缓冲区hash_agg0_sumIsNull true;hash_agg0_sum -1;}isNull$0 in1.isNullAt(0); // 检查第一个字段是否为nullfield$0 -1;if (!isNull$0) {field$0 in1.getInt(0); // 获取第一个字段的整数值}int result$4 -1; // 中间结果boolean isNull$4; // 中间null标记if (isNull$0) {// 处理null的情况isNull$4 hash_agg0_sumIsNull;if (!isNull$4) {result$4 hash_agg0_sum;}} else {int result$3 -1; // 中间结果boolean isNull$3; // 中间null标记if (hash_agg0_sumIsNull) {// 处理聚合和为null的情况isNull$3 isNull$0;if (!isNull$3) {result$3 field$0;}} else {// 计算和isNull$1 hash_agg0_sumIsNull || isNull$0;result$2 -1;if (!isNull$1) {result$2 (int) (hash_agg0_sum field$0); // 将字段值加到聚合和上}isNull$3 isNull$1;if (!isNull$3) {result$3 result$2;}}isNull$4 isNull$3;if (!isNull$4) {result$4 result$3;}}hash_agg0_sumIsNull isNull$4;if (!isNull$4) {// 复制结果值hash_agg0_sum result$4;}}Overridepublic void endInput() throws Exception {int field$0;boolean isNull$0;boolean isNull$1;int result$2;if (hasInput) {if (hash_agg0_sumIsNull) {valueRow$5.setField(0, null); // 如果聚合和为null则将输出字段设置为null} else {valueRow$5.setField(0, hash_agg0_sum); // 将输出字段设置为聚合和}output.collect(outElement.replace(valueRow$5)); // 收集输出元素}}Overridepublic void finish() throws Exception {super.finish(); // 调用父类的finish方法}Overridepublic void close() throws Exception {super.close(); // 调用父类的close方法}
}
