1.10streamSQLkafka关联mysql维表,出现的问题

Đang mở
#325 0 bình luận 0 reaction 0 người được giao Xem trên GitHub

Chưa có ai nhận issue này.

Đánh giá

Độ khó
4/5
Thời gian dự kiến
3-5 ngày
Mức phù hợp với người mới
25/100
Loại issue
Lỗi
Độ rõ ràng
Cần làm rõ
Mức độ hoạt động
Đình trệ
Công nghệ
java, kafka, mysql, sql

Hướng nghiên cứu

Bắt đầu với SQL được cung cấp và stack trace, sau đó đọc com.dtstack.flink.sql.side.JoinNodeDealer.dealJoinNode tại dòng 198 và lần theo các hàm gọi nó trong SideSQLParser.java. Tái hiện truy vấn và xác minh rằng join được xử lý mà không xảy ra SqlJoin-to-SqlBasicCall ClassCastException đã được báo cáo.

Do mô hình lập chỉ mục viết ra từ nội dung của issue.

Mô tả

我的sql
INSERT INTO dwd_business_flow SELECT SUBSTRING(CAST(TIMESTAMPADD(DAY,0,LOCALTIMESTAMP) as varchar),1,10) AS stat_date, CASE WHEN t2.fee_type = 'WAGES' THEN '010' WHEN t2.fee_type = 'RENT' THEN '011' WHEN t2.fee_type = 'DEBIT' THEN '012' WHEN t2.fee_type = 'MEAL' THEN '013' WHEN t2.fee_type = 'REIMBURSE' THEN '015' WHEN t2.fee_type = 'WELFARE' THEN '014' WHEN t2.fee_type = 'RENTBICYLE' THEN '016' WHEN t2.fee_type = 'RENTCLOTHES'THEN '017' WHEN t2.fee_type = 'RENTMOBILE' THEN '019' WHEN t2.fee_type = 'RENTBETTRY' THEN '018' WHEN t2.fee_type = 'RENTCAR' THEN '020' WHEN t2.fee_type = 'LABOUR' THEN '021' WHEN t2.fee_type = 'YEARENDREWARD' THEN '022' WHEN t2.fee_type = 'OTHER' THEN '023' WHEN t2.fee_type = 'SERVICE' THEN '024' ELSE '010' END AS business_type_code, -- 业务类型编号 t2.fee_name AS business_type_name, -- 业务类型名称 t1.order_id AS order_code, -- 订单号 cast(t1.order_detail_no as varchar) AS child_order_code, -- 子订单号 t1.deal_apply_time AS order_create_time, -- 订单创建时间 t1.busi_complete_data AS order_pay_time, -- 订单支付时间 t1.sal_should_pay AS pay_amount, -- 应发金额 t1.sal_real_pay AS pay_amount_real, -- 实发金额 '2' AS trade_state, -- 交易状态(1-未完成,2-已完成) t3.user_code AS user_code, -- 用户编号 t1.user_name, -- 用户名 t1.id_number AS identity_no, -- 身份证号码 t1.mobile, -- 手机号码 t2.corp_no AS company_code, -- 企业编号 t3.company_name AS company_name, -- 企业名称 '1' AS pay_way, -- 支付渠道(1-零钱支付 2-积分支付 3-支付宝支付 4-微信支付) '2' AS trade_type, -- 交易类型(1-个人费用,2-因公费用) '2' AS trade_flow_type, -- 流水类型(1-支出,2-收入) t1.user_info_id AS employee_code, -- 员工主键 t2.relea_month AS sal8_real_month, -- 实际薪酬年月 t1.wallet_acc_id AS trade_account_code, -- 账户编码 case when t4.acct_type = 'DIRECT_RELEASE' then '直发账户' when t4.acct_type = 'WALLET_CUP_NEW' then '好易联账户' when t4.acct_type = 'WALLET_ZSY' then '中顺易账户' when t4.acct_type = 'WALLET_LIMIT' then '额度账户' when t4.acct_type = 'WALLET_FL_PDS' then '华夏PDS账户' else '全部账户' end AS trade_account_name, t1.remark AS remark_, t3.zt_user_code AS user_open_id, -- 用户id t3.zt_employee_code AS staff_open_id, -- 员工id t3.zt_company_code AS company_open_id, -- 企业id t3.company_short_name AS company_short_name FROM (select t1.enable_status, t1.busi_status, t1.busi_complete_data, t1.order_id, t1.wallet_acc_id, t1.id_number, t1.user_info_id, t1.order_detail_no, t1.deal_apply_time, t1.busi_complete_data, t1.sal_should_pay, t1.sal_real_pay, t1.user_name, t1.mobile, t1.remark from to_adw_sal_orders_detail t1 WHERE t1.enable_status = '1' and t1.busi_status = '1' and t1.busi_complete_data >= SUBSTRING(CAST(TIMESTAMPADD(DAY,0,LOCALTIMESTAMP) as varchar),1,10) ) t1 JOIN to_adw_sal_orders t2 ON t1.order_id = t2.busi_proc_id LEFT JOIN (select g.acct_type,g.acct_number from to_adw_sal_orders_detail e left join to_acct_person_account_info g on e.wallet_acc_id = g.acct_number where g.enable_ = '1') t4 ON t1.wallet_acc_id = t4.acct_number LEFT JOIN (select g.client_id, g.identity_no, g.employee_code, g.user_code, g.company_name, g.zt_user_code, g.zt_employee_code, g.zt_company_code, g.company_short_name from to_adw_sal_orders_detail t1 left join dim_zt_user_info g on t1.id_number = g.identity_no where g.client_id = '1ec595e8ba7211e99c7e7cd30ad3a6a8') t3 ON t1.id_number = t3.identity_no and t1.user_info_id = t3.employee_code

报错:
Exception in thread "main" org.apache.flink.client.program.ProgramInvocationException: The main method caused an error: org.apache.calcite.sql.SqlJoin cannot be cast to org.apache.calcite.sql.SqlBasicCall at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:335) at org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:205) at org.apache.flink.client.program.OptimizerPlanEnvironment.getPipeline(OptimizerPlanEnvironment.java:79) at org.apache.flink.client.program.PackagedProgramUtils.getPipelineFromProgram(PackagedProgramUtils.java:101) at org.apache.flink.client.program.PackagedProgramUtils.createJobGraph(PackagedProgramUtils.java:56) at org.apache.flink.client.program.PackagedProgramUtils.createJobGraph(PackagedProgramUtils.java:85) at com.dtstack.flink.sql.launcher.utils.JobGraphBuildUtil.buildJobGraph(JobGraphBuildUtil.java:80) at com.dtstack.flink.sql.launcher.executor.YarnJobClusterExecutor.exec(YarnJobClusterExecutor.java:65) at com.dtstack.flink.sql.launcher.LauncherMain.main(LauncherMain.java:144) Caused by: java.lang.ClassCastException: org.apache.calcite.sql.SqlJoin cannot be cast to org.apache.calcite.sql.SqlBasicCall at com.dtstack.flink.sql.side.JoinNodeDealer.dealJoinNode(JoinNodeDealer.java:198) at com.dtstack.flink.sql.side.SideSQLParser.parseSql(SideSQLParser.java:164) at com.dtstack.flink.sql.side.SideSQLParser.parseSql(SideSQLParser.java:144) at com.dtstack.flink.sql.side.SideSQLParser.parseSql(SideSQLParser.java:137) at com.dtstack.flink.sql.side.SideSQLParser.getExeQueue(SideSQLParser.java:72) at com.dtstack.flink.sql.side.SideSqlExec.exec(SideSqlExec.java:122) at com.dtstack.flink.sql.exec.ExecuteProcessHelper.sqlTranslation(ExecuteProcessHelper.java:223) at com.dtstack.flink.sql.exec.ExecuteProcessHelper.getStreamExecution(ExecuteProcessHelper.java:165) at com.dtstack.flink.sql.Main.main(Main.java:42) 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:321) ... 8 more

Ngôn ngữ chính
Java
Star
2k
Fork
913
Chỉ số merge pull request
Không có pull request nào được merge trong 30 ngày

Hướng dẫn đóng góp

Chưa lập chỉ mục được hướng dẫn đóng góp cho kho mã nguồn này

Bắt đầu từ đâu

  1. Đọc hết issue, rồi đọc hướng dẫn đóng góp của dự án.
  2. Bình luận trên issue rằng bạn sẽ nhận — tránh hai người làm cùng một việc.
  3. Fork repository và làm thay đổi trên một nhánh.
  4. Mở pull request có tham chiếu số hiệu của issue.

Issue khác của DTStack/flinkStreamSQL

Tất cả issue của DTStack/flinkStreamSQL

Issue tương tự

Thêm issue về Java

Nhận issue mới trong hộp thư của bạn

Bản tóm tắt ngắn những issue GitHub phù hợp với người mới.