Postgres (Stable)

Coming soon

Postgre Profiling

PostgreSQL Profiling is a feature that gathers detailed statistics at both the table and column levels within a PostgreSQL database. It is designed to provide insights into the performance and characteristics of the data stored in the database.

1. Table and Column Level Statistics: PostgreSQL Profiling collects statistics not only on tables but also on individual columns within those tables. This granularity allows you to analyze the data in a more detailed manner.

2. SQL-Based Profiler: The profiling mechanism in PostgreSQL is SQL-based, meaning it relies on SQL queries to gather statistics. It doesn't operate in isolation but can be enabled for specific SQL-based data sources within your PostgreSQL environment.

3. Impact on Ingestion Runs: Enabling profiling comes with a trade-off. While it provides valuable insights, it can slow down the process of ingesting data into the database. This slowdown occurs because the profiling process adds an additional workload to the database, which can affect the overall performance of ingestion jobs.

Caution: It's important to exercise caution when utilizing profiling across a large number of tables or extensive datasets, as this can result in substantial expenses. While we have taken steps to optimize the resource usage of the profiler's queries, it is essential for you to carefully consider the specific tables on which profiling is activated and the frequency of profiling runs.

Capabilities Explained:

1. Row and Column Counts:

   • Row Counts: The profiling tool can extract the total number of rows in each table, providing an overview of the data volume.

   • Column Counts: It also captures the count of columns within each table, helping you understand the data structure.

2. Column-Level Information:

   • Null Counts and Proportions: For each column, the tool identifies how many values are null (missing) and calculates the proportion of null values, which is crucial for data quality assessment.

   • Distinct Counts and Proportions: It determines the number of unique values in each column and calculates the proportion of distinct values, offering insights into data cardinality.

   • Statistical Measures: The tool computes various statistical measures for applicable columns, including the minimum, maximum, mean, median, standard deviation, and select quantile values. These statistics reveal the central tendency and variability of the data.

   • Histograms or Frequencies: It can generate histograms or frequency distributions of unique values within columns. These visual representations provide a clearer understanding of data distribution, aiding in data profiling and analysis.

These capabilities enable data professional and compliance team and to comprehensively analyze and profile their data, uncovering valuable insights such as data completeness, data quality, and the distribution of values. This information is essential for tasks like data cleansing, data modeling, and data exploration, ensuring that data-driven decisions are based on a solid understanding of the underlying data.

Supported Sources​

SQL profiling is supported for all SQL sources. Check the individual source page to verify if it supports profiling.

Field	Description
host_port	host URL
databasestring	database (catalog). If set to Null, all databases will be considered for ingestion.
database_aliasstring	[Deprecated] Alias to apply to database when ingesting.
include_table_location_lineageboolean	If the source supports it, include table lineage to the underlying storage location.Default: True
include_tablesboolean	Whether tables should be ingested.Default: True
include_view_lineageboolean	Include table lineage for viewsDefault: False
include_viewsboolean	Whether views should be ingested.Default: True
initial_databasestring	Initial database used to query for the list of databases, when ingesting multiple databases. Note: this is not used if database or sqlalchemy_uri are provided.Default: postgres
optionsobject	Any options specified here will be passed to SQLAlchemy.create_engine as kwargs.
passwordstring(password)	password
platform_instancestring	The instance of the platform that all assets produced by this recipe belong to
schemestring	database schemeDefault: postgresql+psycopg2
sqlalchemy_uristring	URI of database to connect to. See https://docs.sqlalchemy.org/en/14/core/engines.html#database-urls. Takes precedence over other connection parameters.
usernamestring	username
envstring	The environment that all assets produced by this connector belong toDefault: PROD
database_patternAllowDenyPattern	Regex patterns for databases to filter in ingestion. Note: this is not used if database or sqlalchemy_uri are provided.Default: {'allow': ['.*'], 'deny': [], 'ignoreCase': True}
database_pattern.allowarray(string)
database_pattern.denyarray(string)
database_pattern.ignoreCaseboolean	Whether to ignore case sensitivity during pattern matching.Default: True
domainmap(str,AllowDenyPattern)	A class to store allow deny regexes
domain.key.allowarray(string)
domain.key.denyarray(string)
domain.key.ignoreCaseboolean	Whether to ignore case sensitivity during pattern matching.Default: True
profile_patternAllowDenyPattern	Regex patterns to filter tables (or specific columns) for profiling during ingestion. Note that only tables allowed by the table_pattern will be considered.Default: {'allow': ['.*'], 'deny': [], 'ignoreCase': True}
profile_pattern.allowarray(string)
profile_pattern.denyarray(string)
profile_pattern.ignoreCaseboolean	Whether to ignore case sensitivity during pattern matching.Default: True
schema_patternAllowDenyPattern	Default: {'allow': ['.*'], 'deny': ['information_schema'], ...
schema_pattern.allowarray(string)
schema_pattern.denyarray(string)
schema_pattern.ignoreCaseboolean	Whether to ignore case sensitivity during pattern matching.Default: True
table_patternAllowDenyPattern	Regex patterns for tables to filter in ingestion. Specify regex to match the entire table name in database.schema.table format. e.g. to match all tables starting with customer in Customer database and public schema, use the regex 'Customer.public.customer.*'Default: {'allow': ['.*'], 'deny': [], 'ignoreCase': True}
table_pattern.allowarray(string)
table_pattern.denyarray(string)
table_pattern.ignoreCaseboolean	Whether to ignore case sensitivity during pattern matching.Default: True
view_patternAllowDenyPattern	Regex patterns for views to filter in ingestion. Note: Defaults to table_pattern if not specified. Specify regex to match the entire view name in database.schema.view format. e.g. to match all views starting with customer in Customer database and public schema, use the regex 'Customer.public.customer.*'Default: {'allow': ['.*'], 'deny': [], 'ignoreCase': True}
view_pattern.allowarray(string)
view_pattern.denyarray(string)
view_pattern.ignoreCaseboolean	Whether to ignore case sensitivity during pattern matching.Default: True
profilingGEProfilingConfig	Default: {'enabled': False, 'operation_config': {'lower_fre...
profiling.catch_exceptionsboolean	Default: True
profiling.enabledboolean	Whether profiling should be done.Default: False
profiling.field_sample_values_limitinteger	Upper limit for number of sample values to collect for all columns.Default: 20
profiling.include_field_distinct_countboolean	Whether to profile for the number of distinct values for each column.Default: True
profiling.include_field_distinct_value_frequenciesboolean	Whether to profile for distinct value frequencies.Default: False
profiling.include_field_histogramboolean	Whether to profile for the histogram for numeric fields.Default: False
profiling.include_field_max_valueboolean	Whether to profile for the max value of numeric columns.Default: True
profiling.include_field_mean_valueboolean	Whether to profile for the mean value of numeric columns.Default: True
profiling.include_field_median_valueboolean	Whether to profile for the median value of numeric columns.Default: True
profiling.include_field_min_valueboolean	Whether to profile for the min value of numeric columns.Default: True
profiling.include_field_null_countboolean	Whether to profile for the number of nulls for each column.Default: True
profiling.include_field_quantilesboolean	Whether to profile for the quantiles of numeric columns.Default: False
profiling.include_field_sample_valuesboolean	Whether to profile for the sample values for all columns.Default: True
profiling.include_field_stddev_valueboolean	Whether to profile for the standard deviation of numeric columns.Default: True
profiling.limitinteger	Max number of documents to profile. By default, profiles all documents.
profiling.max_number_of_fields_to_profileinteger	A positive integer that specifies the maximum number of columns to profile for any table. None implies all columns. The cost of profiling goes up significantly as the number of columns to profile goes up.
profiling.max_workersinteger	Number of worker threads to use for profiling. Set to 1 to disable.Default: 60
profiling.offsetinteger	Offset in documents to profile. By default, uses no offset.
profiling.partition_datetimestring(date-time)	For partitioned datasets profile only the partition which matches the datetime or profile the latest one if not set. Only Bigquery supports this.
profiling.partition_profiling_enabledboolean	Default: True
profiling.profile_if_updated_since_daysnumber	Profile table only if it has been updated since these many number of days. If set to null, no constraint of last modified time for tables to profile. Supported only in snowflake and BigQuery.
profiling.profile_table_level_onlyboolean	Whether to perform profiling at table-level only, or include column-level profiling as well.Default: False
profiling.profile_table_row_count_estimate_onlyboolean	Use an approximate query for row count. This will be much faster but slightly less accurate. Only supported for Postgres and MySQL.Default: False
profiling.profile_table_row_limitinteger	Profile tables only if their row count is less then specified count. If set to null, no limit on the row count of tables to profile. Supported only in snowflake and BigQueryDefault: 5000000
profiling.profile_table_size_limitinteger	Profile tables only if their size is less then specified GBs. If set to null, no limit on the size of tables to profile. Supported only in snowflake and BigQueryDefault: 5
profiling.query_combiner_enabledboolean	This feature is still experimental and can be disabled if it causes issues. Reduces the total number of queries issued and speeds up profiling by dynamically combining SQL queries where possible.Default: True
profiling.report_dropped_profilesboolean	Whether to report datasets or dataset columns which were not profiled. Set to True for debugging purposes.Default: False
profiling.turn_off_expensive_profiling_metricsboolean	Whether to turn off expensive profiling or not. This turns off profiling for quantiles, distinct_value_frequencies, histogram & sample_values. This also limits maximum number of fields being profiled to 10.Default: False
profiling.operation_configOperationConfig	Experimental feature. To specify operation configs.
profiling.operation_config.lower_freq_profile_enabledboolean	Whether to do profiling at lower freq or not. This does not do any scheduling just adds additional checks to when not to run profiling.Default: False
profiling.operation_config.profile_date_of_monthinteger	Number between 1 to 31 for date of month (both inclusive). If not specified, defaults to Nothing and this field does not take affect.
profiling.operation_config.profile_day_of_weekinteger	Number between 0 to 6 for day of week (both inclusive). 0 is Monday and 6 is Sunday. If not specified, defaults to Nothing and this field does not take affect.
stateful_ingestionStatefulStaleMetadataRemovalConfig	Base specialized config for Stateful Ingestion with stale metadata removal capability.
stateful_ingestion.enabledboolean	The type of the ingestion state provider registered with datahub.Default: False
stateful_ingestion.remove_stale_metadataboolean	Soft-deletes the entities present in the last successful run but missing in the current run with stateful_ingestion enabled.Default: True

Code Coordinates for self ingestion

import logging
from collections import defaultdict
from typing import Any, Dict, Iterable, List, Optional, Tuple, Union

# This import verifies that the dependencies are available.
import psycopg2  # noqa: F401
import sqlalchemy.dialects.postgresql as custom_types

# GeoAlchemy adds support for PostGIS extensions in SQLAlchemy. In order to
# activate it, we must import it so that it can hook into SQLAlchemy. While
# we don't use the Geometry type that we import, we do care about the side
# effects of the import. For more details, see here:
# https://geoalchemy-2.readthedocs.io/en/latest/core_tutorial.html#reflecting-tables.
from geoalchemy2 import Geometry  # noqa: F401
from pydantic import BaseModel
from pydantic.fields import Field
from sqlalchemy import create_engine, inspect
from sqlalchemy.engine.reflection import Inspector

from concur.configuration.common import AllowDenyPattern
from concur.emitter import mce_builder
from concur.emitter.mcp_builder import mcps_from_mce
from concur.ingestion.api.common import PipelineContext
from concur.ingestion.api.decorators import (
    SourceCapability,
    SupportStatus,
    capability,
    config_class,
    platform_name,
    support_status,
)
from concur.ingestion.api.workunit import MetadataWorkUnit
from concur.ingestion.source.sql.sql_common import (
    SQLAlchemySource,
    SqlWorkUnit,
    register_custom_type,
)
from concur.ingestion.source.sql.sql_config import BasicSQLAlchemyConfig
from concur.metadata.com.linkedin.pegasus2avro.schema import (
    ArrayTypeClass,
    BytesTypeClass,
    MapTypeClass,
)

logger: logging.Logger = logging.getLogger(__name__)

register_custom_type(custom_types.ARRAY, ArrayTypeClass)
register_custom_type(custom_types.JSON, BytesTypeClass)
register_custom_type(custom_types.JSONB, BytesTypeClass)
register_custom_type(custom_types.HSTORE, MapTypeClass)


VIEW_LINEAGE_QUERY = """
WITH RECURSIVE view_deps AS (
SELECT DISTINCT dependent_ns.nspname as dependent_schema
, dependent_view.relname as dependent_view
, source_ns.nspname as source_schema
, source_table.relname as source_table
FROM pg_depend
JOIN pg_rewrite ON pg_depend.objid = pg_rewrite.oid
JOIN pg_class as dependent_view ON pg_rewrite.ev_class = dependent_view.oid
JOIN pg_class as source_table ON pg_depend.refobjid = source_table.oid
JOIN pg_namespace dependent_ns ON dependent_ns.oid = dependent_view.relnamespace
JOIN pg_namespace source_ns ON source_ns.oid = source_table.relnamespace
WHERE NOT (dependent_ns.nspname = source_ns.nspname AND dependent_view.relname = source_table.relname)
UNION
SELECT DISTINCT dependent_ns.nspname as dependent_schema
, dependent_view.relname as dependent_view
, source_ns.nspname as source_schema
, source_table.relname as source_table
FROM pg_depend
JOIN pg_rewrite ON pg_depend.objid = pg_rewrite.oid
JOIN pg_class as dependent_view ON pg_rewrite.ev_class = dependent_view.oid
JOIN pg_class as source_table ON pg_depend.refobjid = source_table.oid
JOIN pg_namespace dependent_ns ON dependent_ns.oid = dependent_view.relnamespace
JOIN pg_namespace source_ns ON source_ns.oid = source_table.relnamespace
INNER JOIN view_deps vd
    ON vd.dependent_schema = source_ns.nspname
    AND vd.dependent_view = source_table.relname
    AND NOT (dependent_ns.nspname = vd.dependent_schema AND dependent_view.relname = vd.dependent_view)
)


SELECT source_table, source_schema, dependent_view, dependent_schema
FROM view_deps
WHERE NOT (source_schema = 'information_schema' OR source_schema = 'pg_catalog')
ORDER BY source_schema, source_table;
"""


class ViewLineageEntry(BaseModel):
    # note that the order matches our query above
    # so pydantic is able to parse the tuple using parse_obj
    source_table: str
    source_schema: str
    dependent_view: str
    dependent_schema: str


class BasePostgresConfig(BasicSQLAlchemyConfig):
    scheme = Field(default="postgresql+psycopg2", description="database scheme")
    schema_pattern = Field(default=AllowDenyPattern(deny=["information_schema"]))


class PostgresConfig(BasePostgresConfig):
    include_view_lineage = Field(
        default=False, description="Include table lineage for views"
    )

    database_pattern: AllowDenyPattern = Field(
        default=AllowDenyPattern.allow_all(),
        description=(
            "Regex patterns for databases to filter in ingestion. "
            "Note: this is not used if `database` or `sqlalchemy_uri` are provided."
        ),
    )
    database: Optional[str] = Field(
        default=None,
        description="database (catalog). If set to Null, all databases will be considered for ingestion.",
    )
    initial_database: Optional[str] = Field(
        default="postgres",
        description=(
            "Initial database used to query for the list of databases, when ingesting multiple databases. "
            "Note: this is not used if `database` or `sqlalchemy_uri` are provided."
        ),
    )


@platform_name("Postgres")
@config_class(PostgresConfig)
@support_status(SupportStatus.CERTIFIED)
@capability(SourceCapability.DOMAINS, "Enabled by default")
@capability(SourceCapability.PLATFORM_INSTANCE, "Enabled by default")
@capability(SourceCapability.DATA_PROFILING, "Optionally enabled via configuration")
@capability(SourceCapability.LINEAGE_COARSE, "Optionally enabled via configuration")
class PostgresSource(SQLAlchemySource):
    """
    This plugin extracts the following:

    - Metadata for databases, schemas, views, and tables
    - Column types associated with each table
    - Also supports PostGIS extensions
    - database_alias (optional) can be used to change the name of database to be ingested
    - Table, row, and column statistics via optional SQL profiling
    """

    config: PostgresConfig

    def __init__(self, config: PostgresConfig, ctx: PipelineContext):
        super().__init__(config, ctx, "postgres")

    @classmethod
    def create(cls, config_dict, ctx):
        config = PostgresConfig.parse_obj(config_dict)
        return cls(config, ctx)

    def get_inspectors(self) -> Iterable[Inspector]:
        # Note: get_sql_alchemy_url will choose `sqlalchemy_uri` over the passed in database
        url = self.config.get_sql_alchemy_url(
            database=self.config.database or self.config.initial_database
        )
        logger.debug(f"sql_alchemy_url={url}")
        engine = create_engine(url, **self.config.options)
        with engine.connect() as conn:
            if self.config.database or self.config.sqlalchemy_uri:
                inspector = inspect(conn)
                yield inspector
            else:
                # pg_database catalog -  https://www.postgresql.org/docs/current/catalog-pg-database.html
                # exclude template databases - https://www.postgresql.org/docs/current/manage-ag-templatedbs.html
                databases = conn.execute(
                    "SELECT datname from pg_database where datname not in ('template0', 'template1')"
                )
                for db in databases:
                    if not self.config.database_pattern.allowed(db["datname"]):
                        continue
                    url = self.config.get_sql_alchemy_url(database=db["datname"])
                    with create_engine(url, **self.config.options).connect() as conn:
                        inspector = inspect(conn)
                        yield inspector

    def get_workunits_internal(self) -> Iterable[Union[MetadataWorkUnit, SqlWorkUnit]]:
        yield from super().get_workunits_internal()

        for inspector in self.get_inspectors():
            if self.config.include_view_lineage:
                yield from self._get_view_lineage_workunits(inspector)

    def _get_view_lineage_elements(
        self, inspector: Inspector
    ) -> Dict[Tuple[str, str], List[str]]:
        data: List[ViewLineageEntry] = []
        with inspector.engine.connect() as conn:
            results = conn.execute(VIEW_LINEAGE_QUERY)
            if results.returns_rows is False:
                return {}

            for row in results:
                data.append(ViewLineageEntry.parse_obj(row))

        lineage_elements: Dict[Tuple[str, str], List[str]] = defaultdict(list)
        # Loop over the lineages in the JSON data.
        for lineage in data:
            if not self.config.view_pattern.allowed(lineage.dependent_view):
                self.report.report_dropped(
                    f"{lineage.dependent_schema}.{lineage.dependent_view}"
                )
                continue

            if not self.config.schema_pattern.allowed(lineage.dependent_schema):
                self.report.report_dropped(
                    f"{lineage.dependent_schema}.{lineage.dependent_view}"
                )
                continue

            key = (lineage.dependent_view, lineage.dependent_schema)
            # Append the source table to the list.
            lineage_elements[key].append(
                mce_builder.make_dataset_urn(
                    self.platform,
                    self.get_identifier(
                        schema=lineage.source_schema,
                        entity=lineage.source_table,
                        inspector=inspector,
                    ),
                    self.config.env,
                )
            )

        return lineage_elements

    def _get_view_lineage_workunits(
        self, inspector: Inspector
    ) -> Iterable[MetadataWorkUnit]:
        lineage_elements = self._get_view_lineage_elements(inspector)

        if not lineage_elements:
            return None

        # Loop over the lineage elements dictionary.
        for key, source_tables in lineage_elements.items():
            # Split the key into dependent view and dependent schema
            dependent_view, dependent_schema = key

            # Construct a lineage object.
            urn = mce_builder.make_dataset_urn(
                self.platform,
                self.get_identifier(
                    schema=dependent_schema, entity=dependent_view, inspector=inspector
                ),
                self.config.env,
            )

            # use the mce_builder to ensure that the change proposal inherits
            # the correct defaults for auditHeader and systemMetadata
            lineage_mce = mce_builder.make_lineage_mce(
                source_tables,
                urn,
            )

            for item in mcps_from_mce(lineage_mce):
                yield item.as_workunit()

    def get_identifier(
        self, *, schema: str, entity: str, inspector: Inspector, **kwargs: Any
    ) -> str:
        regular = f"{schema}.{entity}"
        if self.config.database:
            if self.config.database_alias:
                return f"{self.config.database_alias}.{regular}"
            return f"{self.config.database}.{regular}"
        current_database = self.get_db_name(inspector)
        return f"{current_database}.{regular}"

Questions

If you've got any questions on configuring ingestion for Postgres for data mapping, feel free to ping us on our Slack.