PostProcessingFilter

[Lazycatsu]

Hello!
Can we use, somehow, dynamic filter? What I mean. I have dynamic datasources in Spark scripts - I just read csv from directory. What I want - somehow define filter with regex and group all 100500 sources in one also created with regex expression.
Example:
--files--
s3:\my-bucket\my-table\my-table-0001-0000234.gz
s3:\my-bucket\my-table\my-table-0001-0000235.gz
s3:\my-bucket\my-table\my-table-0001-0034505.gz

have to be change:
--datasource--
name: my-table
uri: s3:\my-bucket\my-table

I use: pysprark 3.3.2, spark-3.3-spline-agent-bundle_2.12-2.0.0.jar

[cerveada]

DataSourcePasswordReplacingFilter is doing something very similar, but it also applies the replacements to the params of the data source, not only the URI.

You can copy and modify that filter to do exactly what you want - create a new filter. You just need to package it in jar and provide it to spark so it is available.

[KumarMutt]

i am trying to call producer ap to load lineage data in db so that i can view in spline ui but see below issue

JSON parse error: ; nested exception is com.twitter.finatra.jackson.caseclass.exceptions.CaseClassMappingException: Errors: com.twitter.finatra.jackson.caseclass.exceptions.CaseClassFieldMappingException: attributes.childRefs: requirement failed: Either `__attrId` or `__exprId` should be defined. Was: None, None, com.twitter.finatra.jackson.caseclass.exceptions.CaseClassFieldMappingException: expressions.functions.childRefs: requirement failed: Either `__attrId` or `__exprId` should be defined. Was: None, None

can you plz help on this @wajda

[wajda]

Please attach the JSON you are sending. It seems to be a problem with its structure.
Do you compose it in your custom program or is it produced by the Spline agent?

[KumarMutt]

i compose the json by spline spark agent using HDFSLineageDispatcher

{
    "id": "4d5b6170-3036-58eb-a5fb-0286f701829d",
    "name": "SparkJob_0226202401",
    "labels": {},
    "operations": {
      "write": {
        "outputSource": "file:/home/spline/my-spark-job/data/output/nicknames",
        "append": false,
        "id": "op-0",
        "name": "InsertIntoHadoopFsRelationCommand",
        "childIds": [
          "op-1"
        ],
        "params": {
          "path": "data/output/nicknames"
        },
        "extra": {
          "destinationType": "csv"
        }
      },
      "reads": [
        {
          "inputSources": [
            "file:/home/spline/my-spark-job/data/input/user_favorites.csv"
          ],
          "id": "op-5",
          "name": "LogicalRelation",
          "output": [
            "attr-0",
            "attr-1",
            "attr-2"
          ],
          "params": {
            "header": "true",
            "inferschema": "true",
            "path": "data/input/user_favorites.csv"
          },
          "extra": {
            "sourceType": "csv"
          }
        },
        {
          "inputSources": [
            "file:/home/spline/my-spark-job/data/input/locations.csv"
          ],
          "id": "op-8",
          "name": "LogicalRelation",
          "output": [
            "attr-3",
            "attr-4",
            "attr-5"
          ],
          "params": {
            "header": "true",
            "inferschema": "true",
            "path": "data/input/locations.csv"
          },
          "extra": {
            "sourceType": "csv"
          }
        }
      ],
      "other": [
        {
          "id": "op-4",
          "name": "View",
          "childIds": [
            "op-5"
          ],
          "output": [
            "attr-0",
            "attr-1",
            "attr-2"
          ],
          "params": {
            "desc": "CatalogTable(\nTable: user_favorites\nCreated Time: Tue Feb 20 16:19:34 IST 2024\nLast Access: UNKNOWN\nCreated By: Spark \nType: VIEW\nSchema: root\n |-- id: integer (nullable = true)\n |-- favorite_color: string (nullable = true)\n |-- favorite_city: string (nullable = true)\n)",
            "isTempView": true
          },
          "extra": {}
        },
        {
          "id": "op-7",
          "name": "View",
          "childIds": [
            "op-8"
          ],
          "output": [
            "attr-3",
            "attr-4",
            "attr-5"
          ],
          "params": {
            "desc": "CatalogTable(\nTable: locations\nCreated Time: Tue Feb 20 16:19:35 IST 2024\nLast Access: UNKNOWN\nCreated By: Spark \nType: VIEW\nSchema: root\n |-- id: integer (nullable = true)\n |-- city: string (nullable = true)\n |-- state: string (nullable = true)\n)",
            "isTempView": true
          },
          "extra": {}
        },
        {
          "id": "op-3",
          "name": "SubqueryAlias",
          "childIds": [
            "op-4"
          ],
          "output": [
            "attr-0",
            "attr-1",
            "attr-2"
          ],
          "params": {
            "identifier": "user_favorites"
          },
          "extra": {}
        },
        {
          "id": "op-6",
          "name": "SubqueryAlias",
          "childIds": [
            "op-7"
          ],
          "output": [
            "attr-3",
            "attr-4",
            "attr-5"
          ],
          "params": {
            "identifier": "locations"
          },
          "extra": {}
        },
        {
          "id": "op-2",
          "name": "Join",
          "childIds": [
            "op-3",
            "op-6"
          ],
          "output": [
            "attr-0",
            "attr-1",
            "attr-2",
            "attr-3",
            "attr-4",
            "attr-5"
          ],
          "params": {
            "condition": {
              "id": "expr-0"
            },
            "hint": "",
            "joinType": "INNER"
          },
          "extra": {}
        },
        {
          "id": "op-1",
          "name": "Project",
          "childIds": [
            "op-2"
          ],
          "output": [
            "attr-0",
            "attr-6"
          ],
          "params": {
            "projectList": [
              {
                "id": "attr-0"
              },
              {
                "id": "expr-1"
              }
            ]
          },
          "extra": {}
        }
      ]
    },
    "attributes": [
      {
        "id": "attr-0",
        "dataType": "e63adadc-648a-56a0-9424-3289858cf0bb",
        "childRefs": [],
        "extra": {},
        "name": "id"
      },
      {
        "id": "attr-1",
        "dataType": "75fe27b9-9a00-5c7d-966f-33ba32333133",
        "childRefs": [],
        "extra": {},
        "name": "favorite_color"
      },
      {
        "id": "attr-2",
        "dataType": "75fe27b9-9a00-5c7d-966f-33ba32333133",
        "childRefs": [],
        "extra": {},
        "name": "favorite_city"
      },
      {
        "id": "attr-3",
        "dataType": "e63adadc-648a-56a0-9424-3289858cf0bb",
        "childRefs": [],
        "extra": {},
        "name": "id"
      },
      {
        "id": "attr-4",
        "dataType": "75fe27b9-9a00-5c7d-966f-33ba32333133",
        "childRefs": [],
        "extra": {},
        "name": "city"
      },
      {
        "id": "attr-5",
        "dataType": "75fe27b9-9a00-5c7d-966f-33ba32333133",
        "childRefs": [],
        "extra": {},
        "name": "state"
      },
      {
        "id": "attr-6",
        "dataType": "75fe27b9-9a00-5c7d-966f-33ba32333133",
        "childRefs": [
          {
            "id": "expr-1"
          }
        ],
        "extra": {},
        "name": "nickname"
      }
    ],
    "expressions": {
      "functions": [
        {
          "id": "expr-0",
          "dataType": "a155e715-56ab-59c4-a94b-ed1851a6984a",
          "childRefs": [
            {
              "id": "attr-2"
            },
            {
              "id": "attr-4"
            }
          ],
          "extra": {
            "simpleClassName": "EqualTo",
            "_typeHint": "expr.Binary",
            "symbol": "="
          },
          "name": "equalto",
          "params": {}
        },
        {
          "id": "expr-2",
          "dataType": "75fe27b9-9a00-5c7d-966f-33ba32333133",
          "childRefs": [
            {
              "id": "attr-1"
            },
            {
              "id": "expr-3"
            },
            {
              "id": "attr-5"
            }
          ],
          "extra": {
            "simpleClassName": "Concat",
            "_typeHint": "expr.Generic"
          },
          "name": "concat",
          "params": {}
        },
        {
          "id": "expr-1",
          "dataType": "75fe27b9-9a00-5c7d-966f-33ba32333133",
          "childRefs": [
            {
              "id": "expr-2"
            }
          ],
          "extra": {
            "simpleClassName": "Alias",
            "_typeHint": "expr.Alias"
          },
          "name": "nickname",
          "params": {
            "name": "nickname"
          }
        }
      ],
      "constants": [
        {
          "id": "expr-3",
          "dataType": "ab4da308-91fb-550a-a5e4-beddecff2a2b",
          "extra": {
            "simpleClassName": "Literal",
            "_typeHint": "expr.Literal"
          },
          "value": " "
        }
      ]
    },
    "systemInfo": {
      "name": "spark",
      "version": "3.5.0"
    },
    "agentInfo": {
      "name": "spline",
      "version": "2.0.0"
    },
    "extraInfo": {
      "appName": "SparkJob_0226202401",
      "dataTypes": [
        {
          "id": "e63adadc-648a-56a0-9424-3289858cf0bb",
          "name": "int",
          "nullable": true,
          "_typeHint": "dt.Simple"
        },
        {
          "id": "75fe27b9-9a00-5c7d-966f-33ba32333133",
          "name": "string",
          "nullable": true,
          "_typeHint": "dt.Simple"
        },
        {
          "id": "a155e715-56ab-59c4-a94b-ed1851a6984a",
          "name": "boolean",
          "nullable": true,
          "_typeHint": "dt.Simple"
        },
        {
          "id": "ab4da308-91fb-550a-a5e4-beddecff2a2b",
          "name": "string",
          "nullable": false,
          "_typeHint": "dt.Simple"
        }
      ]
    }
  }
}

@wajda

[KumarMutt]

above json is for execution-plan this i am using while testing http://localhost:8080/producer/execution-plans

however full lineage json data created is like below

{
  "executionPlan": {
    "id": "4d5b6170-3036-58eb-a5fb-0286f701829d",
    "name": "SparkJob_0226202401",
    "labels": {},
    "operations": {
      "write": {
        "outputSource": "file:/home/spline/my-spark-job/data/output/nicknames",
        "append": false,
        "id": "op-0",
        "name": "InsertIntoHadoopFsRelationCommand",
        "childIds": [
          "op-1"
        ],
        "params": {
          "path": "data/output/nicknames"
        },
        "extra": {
          "destinationType": "csv"
        }
      },
      "reads": [
        {
          "inputSources": [
            "file:/home/spline/my-spark-job/data/input/user_favorites.csv"
          ],
          "id": "op-5",
          "name": "LogicalRelation",
          "output": [
            "attr-0",
            "attr-1",
            "attr-2"
          ],
          "params": {
            "header": "true",
            "inferschema": "true",
            "path": "data/input/user_favorites.csv"
          },
          "extra": {
            "sourceType": "csv"
          }
        },
        {
          "inputSources": [
            "file:/home/spline/my-spark-job/data/input/locations.csv"
          ],
          "id": "op-8",
          "name": "LogicalRelation",
          "output": [
            "attr-3",
            "attr-4",
            "attr-5"
          ],
          "params": {
            "header": "true",
            "inferschema": "true",
            "path": "data/input/locations.csv"
          },
          "extra": {
            "sourceType": "csv"
          }
        }
      ],
      "other": [
        {
          "id": "op-4",
          "name": "View",
          "childIds": [
            "op-5"
          ],
          "output": [
            "attr-0",
            "attr-1",
            "attr-2"
          ],
          "params": {
            "desc": "CatalogTable(\nTable: user_favorites\nCreated Time: Tue Feb 20 16:19:34 IST 2024\nLast Access: UNKNOWN\nCreated By: Spark \nType: VIEW\nSchema: root\n |-- id: integer (nullable = true)\n |-- favorite_color: string (nullable = true)\n |-- favorite_city: string (nullable = true)\n)",
            "isTempView": true
          },
          "extra": {}
        },
        {
          "id": "op-7",
          "name": "View",
          "childIds": [
            "op-8"
          ],
          "output": [
            "attr-3",
            "attr-4",
            "attr-5"
          ],
          "params": {
            "desc": "CatalogTable(\nTable: locations\nCreated Time: Tue Feb 20 16:19:35 IST 2024\nLast Access: UNKNOWN\nCreated By: Spark \nType: VIEW\nSchema: root\n |-- id: integer (nullable = true)\n |-- city: string (nullable = true)\n |-- state: string (nullable = true)\n)",
            "isTempView": true
          },
          "extra": {}
        },
        {
          "id": "op-3",
          "name": "SubqueryAlias",
          "childIds": [
            "op-4"
          ],
          "output": [
            "attr-0",
            "attr-1",
            "attr-2"
          ],
          "params": {
            "identifier": "user_favorites"
          },
          "extra": {}
        },
        {
          "id": "op-6",
          "name": "SubqueryAlias",
          "childIds": [
            "op-7"
          ],
          "output": [
            "attr-3",
            "attr-4",
            "attr-5"
          ],
          "params": {
            "identifier": "locations"
          },
          "extra": {}
        },
        {
          "id": "op-2",
          "name": "Join",
          "childIds": [
            "op-3",
            "op-6"
          ],
          "output": [
            "attr-0",
            "attr-1",
            "attr-2",
            "attr-3",
            "attr-4",
            "attr-5"
          ],
          "params": {
            "condition": {
              "id": "expr-0"
            },
            "hint": "",
            "joinType": "INNER"
          },
          "extra": {}
        },
        {
          "id": "op-1",
          "name": "Project",
          "childIds": [
            "op-2"
          ],
          "output": [
            "attr-0",
            "attr-6"
          ],
          "params": {
            "projectList": [
              {
                "id": "attr-0"
              },
              {
                "id": "expr-1"
              }
            ]
          },
          "extra": {}
        }
      ]
    },
    "attributes": [
      {
        "id": "attr-0",
        "dataType": "e63adadc-648a-56a0-9424-3289858cf0bb",
        "childRefs": [],
        "extra": {},
        "name": "id"
      },
      {
        "id": "attr-1",
        "dataType": "75fe27b9-9a00-5c7d-966f-33ba32333133",
        "childRefs": [],
        "extra": {},
        "name": "favorite_color"
      },
      {
        "id": "attr-2",
        "dataType": "75fe27b9-9a00-5c7d-966f-33ba32333133",
        "childRefs": [],
        "extra": {},
        "name": "favorite_city"
      },
      {
        "id": "attr-3",
        "dataType": "e63adadc-648a-56a0-9424-3289858cf0bb",
        "childRefs": [],
        "extra": {},
        "name": "id"
      },
      {
        "id": "attr-4",
        "dataType": "75fe27b9-9a00-5c7d-966f-33ba32333133",
        "childRefs": [],
        "extra": {},
        "name": "city"
      },
      {
        "id": "attr-5",
        "dataType": "75fe27b9-9a00-5c7d-966f-33ba32333133",
        "childRefs": [],
        "extra": {},
        "name": "state"
      },
      {
        "id": "attr-6",
        "dataType": "75fe27b9-9a00-5c7d-966f-33ba32333133",
        "childRefs": [
          {
            "id": "expr-1"
          }
        ],
        "extra": {},
        "name": "nickname"
      }
    ],
    "expressions": {
      "functions": [
        {
          "id": "expr-0",
          "dataType": "a155e715-56ab-59c4-a94b-ed1851a6984a",
          "childRefs": [
            {
              "id": "attr-2"
            },
            {
              "id": "attr-4"
            }
          ],
          "extra": {
            "simpleClassName": "EqualTo",
            "_typeHint": "expr.Binary",
            "symbol": "="
          },
          "name": "equalto",
          "params": {}
        },
        {
          "id": "expr-2",
          "dataType": "75fe27b9-9a00-5c7d-966f-33ba32333133",
          "childRefs": [
            {
              "id": "attr-1"
            },
            {
              "id": "expr-3"
            },
            {
              "id": "attr-5"
            }
          ],
          "extra": {
            "simpleClassName": "Concat",
            "_typeHint": "expr.Generic"
          },
          "name": "concat",
          "params": {}
        },
        {
          "id": "expr-1",
          "dataType": "75fe27b9-9a00-5c7d-966f-33ba32333133",
          "childRefs": [
            {
              "id": "expr-2"
            }
          ],
          "extra": {
            "simpleClassName": "Alias",
            "_typeHint": "expr.Alias"
          },
          "name": "nickname",
          "params": {
            "name": "nickname"
          }
        }
      ],
      "constants": [
        {
          "id": "expr-3",
          "dataType": "ab4da308-91fb-550a-a5e4-beddecff2a2b",
          "extra": {
            "simpleClassName": "Literal",
            "_typeHint": "expr.Literal"
          },
          "value": " "
        }
      ]
    },
    "systemInfo": {
      "name": "spark",
      "version": "3.5.0"
    },
    "agentInfo": {
      "name": "spline",
      "version": "2.0.0"
    },
    "extraInfo": {
      "appName": "SparkJob_0226202401",
      "dataTypes": [
        {
          "id": "e63adadc-648a-56a0-9424-3289858cf0bb",
          "name": "int",
          "nullable": true,
          "_typeHint": "dt.Simple"
        },
        {
          "id": "75fe27b9-9a00-5c7d-966f-33ba32333133",
          "name": "string",
          "nullable": true,
          "_typeHint": "dt.Simple"
        },
        {
          "id": "a155e715-56ab-59c4-a94b-ed1851a6984a",
          "name": "boolean",
          "nullable": true,
          "_typeHint": "dt.Simple"
        },
        {
          "id": "ab4da308-91fb-550a-a5e4-beddecff2a2b",
          "name": "string",
          "nullable": false,
          "_typeHint": "dt.Simple"
        }
      ]
    }
  },
  "executionEvent": {
    "planId": "4d5b6170-3036-58eb-a5fb-0286f701829d",
    "labels": {},
    "timestamp": 1708426179061,
    "durationNs": 1260907651,
    "extra": {
      "appId": "local-1708426115673",
      "user": "spline",
      "readMetrics": {},
      "writeMetrics": {}
    }
  }
}

@wajda

[KumarMutt]

@wajda can you plz help on this?