[QUESTION] lineage event don't output when spark sql execute fail

[smlHao]

Code of Conduct

• I agree to follow this project's Code of Conduct

Search before asking

• I have searched in the issues and found no similar issues.

What would you like to be improved?

using the KYUUBI_EVENT dispatcher and custom EventHandler to send sql lineage to kafka, when spark sql execute fail, the lineage event output as below, the message is too long ，is there a way to avoid the exception output when spark sql execute fail ?

{
    "executionId": 3,
    "eventTime": 1696903918899,
    "lineage": null,
    "exception": {
        "message": "Table default.tt21 already exists. You need to drop it first.",
        "line": null,
        "startPosition": null,
        "cause": null,
        "errorClass": null,
        "messageParameters": [],
        "stackTrace": [
            {
                "classLoaderName": "app",
                "moduleName": null,
                "moduleVersion": null,
                "methodName": "tableAlreadyExistsError",
                "fileName": "QueryCompilationErrors.scala",
                "lineNumber": 1773,
                "className": "org.apache.spark.sql.errors.QueryCompilationErrors$",
                "nativeMethod": false
            },
            {
                "classLoaderName": "app",
                "moduleName": null,
                "moduleVersion": null,
                "methodName": "run",
                "fileName": "createDataSourceTables.scala",
                "lineNumber": 161,
                "className": "org.apache.spark.sql.execution.command.CreateDataSourceTableAsSelectCommand",
                "nativeMethod": false
            },
            {
                "classLoaderName": "app",
                "moduleName": null,
                "moduleVersion": null,
                "methodName": "sideEffectResult$lzycompute",
                "fileName": "commands.scala",
                "lineNumber": 113,
                "className": "org.apache.spark.sql.execution.command.DataWritingCommandExec",
                "nativeMethod": false
            },
            {
                "classLoaderName": "app",
                "moduleName": null,
                "moduleVersion": null,
                "methodName": "sideEffectResult",
                "fileName": "commands.scala",
                "lineNumber": 111,
                "className": "org.apache.spark.sql.execution.command.DataWritingCommandExec",
                "nativeMethod": false
            },
            {
                "classLoaderName": "app",
                "moduleName": null,
                "moduleVersion": null,
                "methodName": "executeCollect",
                "fileName": "commands.scala",
                "lineNumber": 125,
                "className": "org.apache.spark.sql.execution.command.DataWritingCommandExec",
                "nativeMethod": false
            },
            {
                "classLoaderName": "app",
                "moduleName": null,
                "moduleVersion": null,
                "methodName": "$anonfun$applyOrElse$1",
                "fileName": "QueryExecution.scala",
                "lineNumber": 98,
                "className": "org.apache.spark.sql.execution.QueryExecution$$anonfun$eagerlyExecuteCommands$1",
                "nativeMethod": false
            },
            {
                "classLoaderName": "app",
                "moduleName": null,
                "moduleVersion": null,
                "methodName": "$anonfun$withNewExecutionId$6",
                "fileName": "SQLExecution.scala",
                "lineNumber": 109,
                "className": "org.apache.spark.sql.execution.SQLExecution$",
                "nativeMethod": false
            },
            {
                "classLoaderName": "app",
                "moduleName": null,
                "moduleVersion": null,
                "methodName": "withSQLConfPropagated",
                "fileName": "SQLExecution.scala",
                "lineNumber": 169,
                "className": "org.apache.spark.sql.execution.SQLExecution$",
                "nativeMethod": false
            },
            {
                "classLoaderName": "app",
                "moduleName": null,
                "moduleVersion": null,
                "methodName": "$anonfun$withNewExecutionId$1",
                "fileName": "SQLExecution.scala",
                "lineNumber": 95,
                "className": "org.apache.spark.sql.execution.SQLExecution$",
                "nativeMethod": false
            },
            {
                "classLoaderName": "app",
                "moduleName": null,
                "moduleVersion": null,
                "methodName": "withActive",
                "fileName": "SparkSession.scala",
                "lineNumber": 779,
                "className": "org.apache.spark.sql.SparkSession",
                "nativeMethod": false
            },
            {
                "classLoaderName": "app",
                "moduleName": null,
                "moduleVersion": null,
                "methodName": "withNewExecutionId",
                "fileName": "SQLExecution.scala",
                "lineNumber": 64,
                "className": "org.apache.spark.sql.execution.SQLExecution$",
                "nativeMethod": false
            },
            {
                "classLoaderName": "app",
                "moduleName": null,
                "moduleVersion": null,
                "methodName": "applyOrElse",
                "fileName": "QueryExecution.scala",
                "lineNumber": 98,
                "className": "org.apache.spark.sql.execution.QueryExecution$$anonfun$eagerlyExecuteCommands$1",
                "nativeMethod": false
            },
            {
                "classLoaderName": "app",
                "moduleName": null,
                "moduleVersion": null,
                "methodName": "applyOrElse",
                "fileName": "QueryExecution.scala",
                "lineNumber": 94,
                "className": "org.apache.spark.sql.execution.QueryExecution$$anonfun$eagerlyExecuteCommands$1",
                "nativeMethod": false
            },
            {
                "classLoaderName": "app",
                "moduleName": null,
                "moduleVersion": null,
                "methodName": "$anonfun$transformDownWithPruning$1",
                "fileName": "TreeNode.scala",
                "lineNumber": 584,
                "className": "org.apache.spark.sql.catalyst.trees.TreeNode",
                "nativeMethod": false
            },
            {
                "classLoaderName": "app",
                "moduleName": null,
                "moduleVersion": null,
                "methodName": "withOrigin",
                "fileName": "TreeNode.scala",
                "lineNumber": 176,
                "className": "org.apache.spark.sql.catalyst.trees.CurrentOrigin$",
                "nativeMethod": false
            },
            {
                "classLoaderName": "app",
                "moduleName": null,
                "moduleVersion": null,
                "methodName": "transformDownWithPruning",
                "fileName": "TreeNode.scala",
                "lineNumber": 584,
                "className": "org.apache.spark.sql.catalyst.trees.TreeNode",
                "nativeMethod": false
            },
            {
                "classLoaderName": "app",
                "moduleName": null,
                "moduleVersion": null,
                "methodName": "org$apache$spark$sql$catalyst$plans$logical$AnalysisHelper$$super$transformDownWithPruning",
                "fileName": "LogicalPlan.scala",
                "lineNumber": 30,
                "className": "org.apache.spark.sql.catalyst.plans.logical.LogicalPlan",
                "nativeMethod": false
            },
            {
                "classLoaderName": "app",
                "moduleName": null,
                "moduleVersion": null,
                "methodName": "transformDownWithPruning",
                "fileName": "AnalysisHelper.scala",
                "lineNumber": 267,
                "className": "org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper",
                "nativeMethod": false
            },
            {
                "classLoaderName": "app",
                "moduleName": null,
                "moduleVersion": null,
                "methodName": "transformDownWithPruning$",
                "fileName": "AnalysisHelper.scala",
                "lineNumber": 263,
                "className": "org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper",
                "nativeMethod": false
            },
            {
                "classLoaderName": "app",
                "moduleName": null,
                "moduleVersion": null,
                "methodName": "transformDownWithPruning",
                "fileName": "LogicalPlan.scala",
                "lineNumber": 30,
                "className": "org.apache.spark.sql.catalyst.plans.logical.LogicalPlan",
                "nativeMethod": false
            },
            {
                "classLoaderName": "app",
                "moduleName": null,
                "moduleVersion": null,
                "methodName": "transformDownWithPruning",
                "fileName": "LogicalPlan.scala",
                "lineNumber": 30,
                "className": "org.apache.spark.sql.catalyst.plans.logical.LogicalPlan",
                "nativeMethod": false
            },
            {
                "classLoaderName": "app",
                "moduleName": null,
                "moduleVersion": null,
                "methodName": "transformDown",
                "fileName": "TreeNode.scala",
                "lineNumber": 560,
                "className": "org.apache.spark.sql.catalyst.trees.TreeNode",
                "nativeMethod": false
            },
            {
                "classLoaderName": "app",
                "moduleName": null,
                "moduleVersion": null,
                "methodName": "eagerlyExecuteCommands",
                "fileName": "QueryExecution.scala",
                "lineNumber": 94,
                "className": "org.apache.spark.sql.execution.QueryExecution",
                "nativeMethod": false
            },
            {
                "classLoaderName": "app",
                "moduleName": null,
                "moduleVersion": null,
                "methodName": "commandExecuted$lzycompute",
                "fileName": "QueryExecution.scala",
                "lineNumber": 81,
                "className": "org.apache.spark.sql.execution.QueryExecution",
                "nativeMethod": false
            },
            {
                "classLoaderName": "app",
                "moduleName": null,
                "moduleVersion": null,
                "methodName": "commandExecuted",
                "fileName": "QueryExecution.scala",
                "lineNumber": 79,
                "className": "org.apache.spark.sql.execution.QueryExecution",
                "nativeMethod": false
            },
            {
                "classLoaderName": "app",
                "moduleName": null,
                "moduleVersion": null,
                "methodName": "<init>",
                "fileName": "Dataset.scala",
                "lineNumber": 220,
                "className": "org.apache.spark.sql.Dataset",
                "nativeMethod": false
            },
            {
                "classLoaderName": "app",
                "moduleName": null,
                "moduleVersion": null,
                "methodName": "$anonfun$ofRows$2",
                "fileName": "Dataset.scala",
                "lineNumber": 100,
                "className": "org.apache.spark.sql.Dataset$",
                "nativeMethod": false
            },
            {
                "classLoaderName": "app",
                "moduleName": null,
                "moduleVersion": null,
                "methodName": "withActive",
                "fileName": "SparkSession.scala",
                "lineNumber": 779,
                "className": "org.apache.spark.sql.SparkSession",
                "nativeMethod": false
            },
            {
                "classLoaderName": "app",
                "moduleName": null,
                "moduleVersion": null,
                "methodName": "ofRows",
                "fileName": "Dataset.scala",
                "lineNumber": 97,
                "className": "org.apache.spark.sql.Dataset$",
                "nativeMethod": false
            },
            {
                "classLoaderName": "app",
                "moduleName": null,
                "moduleVersion": null,
                "methodName": "$anonfun$sql$1",
                "fileName": "SparkSession.scala",
                "lineNumber": 622,
                "className": "org.apache.spark.sql.SparkSession",
                "nativeMethod": false
            },
            {
                "classLoaderName": "app",
                "moduleName": null,
                "moduleVersion": null,
                "methodName": "withActive",
                "fileName": "SparkSession.scala",
                "lineNumber": 779,
                "className": "org.apache.spark.sql.SparkSession",
                "nativeMethod": false
            },
            {
                "classLoaderName": "app",
                "moduleName": null,
                "moduleVersion": null,
                "methodName": "sql",
                "fileName": "SparkSession.scala",
                "lineNumber": 617,
                "className": "org.apache.spark.sql.SparkSession",
                "nativeMethod": false
            },
            {
                "classLoaderName": null,
                "moduleName": null,
                "moduleVersion": null,
                "methodName": "$anonfun$executeStatement$1",
                "fileName": "ExecuteStatement.scala",
                "lineNumber": 86,
                "className": "org.apache.kyuubi.engine.spark.operation.ExecuteStatement",
                "nativeMethod": false
            },
            {
                "classLoaderName": "app",
                "moduleName": null,
                "moduleVersion": null,
                "methodName": "apply",
                "fileName": "JFunction0$mcV$sp.java",
                "lineNumber": 23,
                "className": "scala.runtime.java8.JFunction0$mcV$sp",
                "nativeMethod": false
            },
            {
                "classLoaderName": null,
                "moduleName": null,
                "moduleVersion": null,
                "methodName": "$anonfun$withLocalProperties$1",
                "fileName": "SparkOperation.scala",
                "lineNumber": 155,
                "className": "org.apache.kyuubi.engine.spark.operation.SparkOperation",
                "nativeMethod": false
            },
            {
                "classLoaderName": "app",
                "moduleName": null,
                "moduleVersion": null,
                "methodName": "withSQLConfPropagated",
                "fileName": "SQLExecution.scala",
                "lineNumber": 169,
                "className": "org.apache.spark.sql.execution.SQLExecution$",
                "nativeMethod": false
            },
            {
                "classLoaderName": null,
                "moduleName": null,
                "moduleVersion": null,
                "methodName": "withLocalProperties",
                "fileName": "SparkOperation.scala",
                "lineNumber": 139,
                "className": "org.apache.kyuubi.engine.spark.operation.SparkOperation",
                "nativeMethod": false
            },
            {
                "classLoaderName": null,
                "moduleName": null,
                "moduleVersion": null,
                "methodName": "executeStatement",
                "fileName": "ExecuteStatement.scala",
                "lineNumber": 81,
                "className": "org.apache.kyuubi.engine.spark.operation.ExecuteStatement",
                "nativeMethod": false
            },
            {
                "classLoaderName": null,
                "moduleName": null,
                "moduleVersion": null,
                "methodName": "run",
                "fileName": "ExecuteStatement.scala",
                "lineNumber": 103,
                "className": "org.apache.kyuubi.engine.spark.operation.ExecuteStatement$$anon$1",
                "nativeMethod": false
            },
            {
                "classLoaderName": null,
                "moduleName": "java.base",
                "moduleVersion": "11.0.13",
                "methodName": "call",
                "fileName": null,
                "lineNumber": -1,
                "className": "java.util.concurrent.Executors$RunnableAdapter",
                "nativeMethod": false
            },
            {
                "classLoaderName": null,
                "moduleName": "java.base",
                "moduleVersion": "11.0.13",
                "methodName": "run",
                "fileName": null,
                "lineNumber": -1,
                "className": "java.util.concurrent.FutureTask",
                "nativeMethod": false
            },
            {
                "classLoaderName": null,
                "moduleName": "java.base",
                "moduleVersion": "11.0.13",
                "methodName": "runWorker",
                "fileName": null,
                "lineNumber": -1,
                "className": "java.util.concurrent.ThreadPoolExecutor",
                "nativeMethod": false
            },
            {
                "classLoaderName": null,
                "moduleName": "java.base",
                "moduleVersion": "11.0.13",
                "methodName": "run",
                "fileName": null,
                "lineNumber": -1,
                "className": "java.util.concurrent.ThreadPoolExecutor$Worker",
                "nativeMethod": false
            },
            {
                "classLoaderName": null,
                "moduleName": "java.base",
                "moduleVersion": "11.0.13",
                "methodName": "run",
                "fileName": null,
                "lineNumber": -1,
                "className": "java.lang.Thread",
                "nativeMethod": false
            }
        ],
        "sqlState": null,
        "internalError": false,
        "simpleMessage": "Table default.tt21 already exists. You need to drop it first.",
        "suppressed": [],
        "localizedMessage": "Table default.tt21 already exists. You need to drop it first."
    },
    "eventType": "operation_lineage_kyuubi"
}

How should we improve?

No response

Are you willing to submit PR?

• Yes. I would be willing to submit a PR with guidance from the Kyuubi community to improve.
• No. I cannot submit a PR at this time.

[github-actions[bot]]

Hello @smlHao,
Thanks for finding the time to report the issue!
We really appreciate the community's efforts to improve Apache Kyuubi.

[pan3793]

The EventHandler allows you to do anything in the apply method, you can drop the exception information before serializing the event to JSON

[smlHao]

thanks ！！！ in the apply method ， add an if clause when lineage is null don't send ,works fine

[wForget]

We might be able to add a Serializer(configurable) for org.apache.kyuubi.events.JsonProtocol#mapper to serialize exception object to a pretty string.

[smlHao]

@wForget I found that when a sql, will send two OperationLineageKyuubiEvent records and every OperationLineageKyuubiEvent record's executionId is different, is it right ?

[wForget]

This is expected behavior for dml statements, you can clearly see in the sql tab of spark ui that there are actually two executions, one by spark.sql() and one by result.collect().

[smlHao]

thanks !!! if don't want to send two lineage event, should code to avoid this in the EventHandler's apply method ?

[wForget]

I think it's better in SparkOperationLineageQueryExecutionListener or KyuubiEventDispatcher, only for lineage events.

[smlHao]

thanks !!! got it