DataAI Spark Function Catalog

Add the functions artifact

<dependency>
  <groupId>com.dataai</groupId>
  <artifactId>dataai-spark-functions</artifactId>
  <version>0.1.0-SNAPSHOT</version>
</dependency>

Install all modules locally during source development:

Set-Location C:\Projects\DataAI.Etl\spark
mvn clean install

Function families

ETL and quality

Declarative rule types:

• REQUIRED
• UNIQUE
• MINIMUM
• MAXIMUM
• BETWEEN
• IN_SET
• DATE_FORMAT
• LENGTH
• EQUALS
• REGEX

DataAiResult result = DataAiPipeline.fromDataset(input)
        .normalize()
        .recordKey("order_id")
        .profile()
        .validate(
                RuleSpec.required("customer-required", "customer_id"),
                RuleSpec.unique("order-unique", "order_id"),
                RuleSpec.between("amount-range", "amount", 0, 100000),
                RuleSpec.inSet("status-values", "status", List.of("Open", "Closed")),
                RuleSpec.dateFormat("order-date", "order_date", "yyyy-MM-dd"),
                RuleSpec.length("reference-length", "reference", 1, 40))
        .execute();

General analytics

Supported aggregations: COUNT, COUNT_DISTINCT, SUM, MINIMUM, MAXIMUM, AVERAGE, STANDARD_DEVIATION.

Dataset<Row> variance = AnalyticsFunctions.variance(
        input,
        List.of("region"),
        "period",
        "2025",
        "2026",
        "revenue",
        Aggregation.SUM);

Dataset<Row> regression = AnalyticsFunctions.linearRegression(
        input, "units", "revenue", "region", 100.0);

Time analysis

Dataset<Row> rolling = TimeSeriesFunctions.rolling(
        input,
        List.of("region"),
        "order_date",
        TimePeriod.MONTH,
        "revenue",
        Aggregation.SUM,
        3,
        RollingOperation.MOVING_AVERAGE);

Business analytics

Dataset<Row> pareto = BusinessFunctions.pareto(
        input, "product", "revenue", Aggregation.SUM);

Dataset<Row> cohorts = BusinessFunctions.cohort(
        input, "customer_id", "activity_date", "revenue", TimePeriod.MONTH);

Market models

Assumption arguments use percentage points: 10.0 means 10 percent.

Dataset<Row> demand = MarketFunctions.demand(
        input,
        List.of("region", "product"),
        "units",
        "order_date",
        TimePeriod.MONTH,
        10.0);

Dataset<Row> profit = MarketFunctions.profit(
        input,
        List.of("region", "product"),
        "revenue",
        "direct_cost",
        null,
        null,
        0);

Geographic analysis

MapFunctions.readiness(...) returns a MapReadinessResult containing:

• classified source rows;
• a readiness summary;
• valid-coordinate rows;
• invalid-coordinate rows;
• suggested latitude fields;
• suggested longitude fields.

Checks classify missing coordinates, invalid latitude/longitude ranges, duplicate locations, valid coordinates without a name, and KML-ready rows.

MapReadinessResult map = MapFunctions.readiness(
        input, "latitude", "longitude", "location_name");

Matrix analysis and balancing

MatrixFunctions.crossTab(...) produces a Spark pivot. balance(...) performs iterative proportional fitting against row and column control totals.

MatrixBalanceResult balanced = MatrixFunctions.balance(
        cells,
        "region",
        "category",
        "value",
        rowTargets,
        columnTargets,
        "target_total",
        50,
        0.001);

if (!balanced.converged()) {
    throw new IllegalStateException("Matrix controls did not converge");
}

The balanced output includes original and balanced values, balancing coefficients, row/column targets, final totals, differences, and convergence metadata.

Data dictionary, recommendations, alerts, and narratives

Narratives are deterministic and local. They do not send customer data to an external AI endpoint. Customers can optionally send the returned, governed summary DataFrame to their own approved AI integration.

Platform output adapters

Tableau output adapter

Add com.dataai:dataai-spark-tableau:0.1.0-SNAPSHOT when a Spark application needs standardized DataAI outputs for Tableau. TableauOutputs.from(...) accepts a DataAiResult and returns a TableauOutputBundle containing:

• cleanRows() and rejectedRows();
• qualityFindings() and fieldProfiles();
• a one-row pipelineRuns() DataFrame;
• a one-row dashboardMetrics() DataFrame with row totals, quality score, finding counts by severity, profile count, and total null values.

TableauOutputNames provides stable default table-name constants. The adapter does not persist, cache, collect, connect to Tableau, or make network calls. The customer explicitly writes the returned DataFrames to a format/catalog visible through Tableau's native Spark SQL or Databricks connector.

Use TableauFunctionOutputs.withRunMetadata(...) with any function-result DataFrame to add reserved result name, run ID, completion time, and library version fields. Use TableauFunctionOutputs.matrixBalance(...) to convert a MatrixBalanceResult with iteration count, maximum error, and convergence fields.

TableauOutputBundle tableau = TableauOutputs.from(result);

tableau.dashboardMetrics().write()
        .format("delta")
        .mode(SaveMode.Append)
        .saveAsTable("analytics.dataai_dashboard_metrics");

See Tableau output schema and function-to-Tableau coverage.

InterSystems IRIS adapter

Add com.dataai:dataai-spark-iris:0.1.0-SNAPSHOT when DataAI must read from or write approved outputs to InterSystems IRIS. IrisJdbcOptions creates a safe, customer-controlled JDBC configuration; IrisDataFrames provides table, partitioned-table, query, and explicitly invoked writer entry points.

IrisPipelineOutputs.from(...) converts a DataAiResult into clean, rejected, finding, profile, and pipeline-run DataFrames with stable IRIS output names. The adapter never chooses a target table or save mode and never invokes save() for the customer.

Use IrisFunctionOutputs.withRunMetadata(...) for any result DataFrame. Use IrisFunctionOutputs.matrixBalance(...) for matrix balancing with run, library, platform, convergence, iteration, and maximum-error metadata.

See IRIS function outputs and IRIS README. The adapter uses Spark's standard JDBC source and does not bundle the InterSystems driver, persist automatically, collect full data, call a DataAI service, or emit telemetry.

Execution behavior

Most APIs return lazy Spark DataFrames. Spark executes them only when a caller runs an action or writes the result. Two APIs intentionally perform actions:

• AnalyticsFunctions.correlations(...) calculates one Spark correlation for each requested field pair and builds a small result DataFrame on the driver.
• MatrixFunctions.balance(...) evaluates convergence after each balancing iteration.

Keep correlation field lists bounded and configure matrix iterations and tolerance appropriate to the data size.

Capabilities delegated to the customer's platform

The following DataAI application features are not copied into the computation JAR because Spark or the customer's BI/orchestration platform already owns them:

This boundary keeps DataAI ETL an embeddable library rather than turning it into a hosted service.