Set up the Openflow Connector for Microsoft Dataverse¶

This topic describes the steps to set up the Openflow Connector for Microsoft Dataverse.

Prerequisites¶

1. Ensure that you have reviewed About Openflow Connector for Microsoft Dataverse.
2. Ensure that you have Set up Openflow - BYOC or Set up Openflow - Snowflake Deployments.
3. If using Openflow - Snowflake Deployments, ensure that you’ve reviewed configuring required domains and have granted access to the required domains for the Microsoft Dataverse connector.

Get the credentials¶

As a Microsoft Dataverse administrator, perform the following steps:

1. Ensure you have a Dataverse Environment to work with, and you have access to that environment through https://admin.powerplatform.microsoft.com/.
2. Ensure that you have an application registered in Microsoft Entra ID in portal.azure.com. This application must have access to the tenant we have our Dataverse Environment available. To register the application follow this guide.
3. Generate and store ClientID and Client Secret within that application.
4. Go to Power Apps Admin Center and configure your Dataverse Environment to be accessed via applications registered before. To do that, go to Manage » Environments and select the environment to configure. Then go to Settings » Users & permissions » Application users. Previously created applications must be added and granted with privileges necessary to read data from Microsoft Dataverse.
5. Copy and save the Environment URL of the selected Dataverse Environment from https://admin.powerplatform.microsoft.com/.

Set up Snowflake account¶

As a Snowflake account administrator, perform the following tasks:

1. Create a Snowflake user with the type as SERVICE.

   Create a database and schema to store the replicated data, and set up privileges for the service user to create tables in destination schema by granting the USAGE and CREATE TABLE privileges.

   Copy code

   CREATE DATABASE <destination_database>;
   CREATE SCHEMA <destination_database>.<destination_schema>;
   CREATE USER <openflow_user> TYPE=SERVICE COMMENT='Service user for automated access of Openflow';
   CREATE ROLE <openflow_role>;
   GRANT ROLE <openflow_role> TO USER <openflow_user>;
   GRANT USAGE ON DATABASE <destination_database> TO ROLE <openflow_role>;
   GRANT USAGE ON SCHEMA <destination_database>.<destination_schema> TO ROLE <openflow_role>;
   GRANT CREATE TABLE ON SCHEMA <destination_database>.<destination_schema> TO ROLE <openflow_role>;
   CREATE WAREHOUSE <openflow_warehouse>
     WITH
         WAREHOUSE_SIZE = 'SMALL'
         AUTO_SUSPEND = 300
         AUTO_RESUME = TRUE;
   GRANT USAGE, OPERATE ON WAREHOUSE <openflow_warehouse> TO ROLE <openflow_role>;
   

   1. Create a pair of secure keys (public and private). Store the private key for the user in a file to supply to the connector’s configuration. Assign the public key to the Snowflake service user:

      Copy code

      ALTER USER <openflow_user> SET RSA_PUBLIC_KEY = 'thekey';
      

      For more information, see pair of keys.

2. Snowflake strongly recommends this step. Configure a secrets manager supported by Openflow, for example, AWS, Azure, and Hashicorp, and store the public and private keys in the secret store.

   Note

   If for any reason, you do not wish to use a secrets manager, then you are responsible for safeguarding the public key and private key files used for key-pair authentication according to the security policies of your organization.

   1. Once the secrets manager is configured, determine how you will authenticate to it. On AWS, it’s recommended that you the EC2 instance role associated with Openflow as this way no other secrets have to be persisted.
   2. In Openflow, configure a Parameter Provider associated with this Secrets Manager, from the hamburger menu in the upper right. Navigate to Controller Settings » Parameter Provider and then fetch your parameter values.
   3. At this point all credentials can be referenced with the associated parameter paths and no sensitive values need to be persisted within Openflow.

3. If any other Snowflake users require access to the raw ingested documents and tables ingested by the connector (for example, for custom processing in Snowflake), then grant those users the role created in step 1.

4. Designate a warehouse for the connector to use. Grant the USAGE privilege on the warehouse to the role created before. Start with the smallest warehouse size, then experiment with size depending on the number of tables being replicated, and the amount of data transferred. Large table numbers typically scale better with multi-cluster warehouses, rather than larger warehouse sizes.

Set up the connector¶

As a data engineer, perform the following tasks to install and configure the connector:

Install the connector¶

Configure the connector¶

1. Right-click on the imported process group and select Parameters.
2. Populate the required parameter values as described in Flow parameters.

Flow parameters¶

This section describes the flow parameters that you can configure based on the following parameter contexts:

• Dataverse Source Parameters: Used to establish connection with Dataverse.
• Dataverse Destination Parameters: Used to establish connection with Snowflake.
• Dataverse Ingestion Parameters: Used to define the configuration of data downloaded from Dataverse.

Dataverse Source Parameters¶

Dataverse Destination Parameters¶

Dataverse Ingestion Parameters¶

Run the flow¶

1. Right-click on the plane and select Enable all Controller Services.
2. Right-click on the imported process group and select Start. The connector starts the data ingestion.

Replicate a subset of columns in a table¶

The connector can filter the data replicated per table to a subset of configured columns.

To apply filters to columns, set the Column Filter JSON parameter in the Dataverse ingestion parameters to a JSON filter. Add an array of configurations, one entry for every table to which you want to apply a filter.

The following example excludes large binary columns from a table:

[
    {
        "table": "mytable",
        "excluded": ["mycolumn", "mycolumn_binary"]
    }
]

Columns can be included or excluded by name or pattern. You can apply a single condition per table, or combine multiple conditions, with exclusions taking precedence over inclusions.

The following example shows all available fields. The table field is mandatory. One or more of included, excluded, includedPattern, excludedPattern is required.

[
    {
        "table" : "<singular logical entity name>",
        "included": ["<column name>", "<column name>"],
        "excluded": ["<column name>", "<column name>"],
        "includedPattern": "<regular expression>",
        "excludedPattern": "<regular expression>",
    }
]

Include formatted values¶

Microsoft Dataverse stores many columns as raw values, such as an option set integer, a lookup GUID, or an unformatted amount. For those columns, Dataverse can also return a display value, called a formatted value: the localized label of a choice, yes/no, status, or status reason column, the primary name of a lookup or owner column, and locale-formatted numbers, currencies, and dates. For details about which columns have formatted values, see Select columns using OData in the Microsoft documentation.

By default, the connector replicates raw values only. To also replicate formatted values, set the Include Formatted Values parameter in the Dataverse ingestion parameters to true, then restart the connector process group so that the change takes effect.

Raw columns are unchanged when you enable this parameter. Instead, the connector adds a single _DATAVERSE_FORMATTED_VALUES column of type OBJECT to each replicated table. The keys of the object are source column names and the values are always strings:

{
  "statuscode": "Active",
  "annualincome": "$80,000.00",
  "_transactioncurrencyid_value": "US Dollar"
}

Keep the following in mind when you query the column:

• Lookup and owner columns keep the _<column>_value form that Dataverse uses, as shown by _transactioncurrencyid_value in the preceding example.
• Columns that don’t have a formatted value are absent from the object. If a row has no formatted values at all, _DATAVERSE_FORMATTED_VALUES is NULL for that row.
• Rows that were deleted at the source don’t carry formatted values. For more information about how deletes are replicated, see Connector-managed columns.
• If you enable the parameter for a connector that already replicated data, the column is populated from the next ingestion onwards. Rows that were replicated earlier keep NULL until they change at the source again.
• If a source table already has a column named _DATAVERSE_FORMATTED_VALUES, ingestion of that table fails.

To read a single formatted value, extract it by key and cast it:

SELECT
    "statuscode",
    _DATAVERSE_FORMATTED_VALUES:statuscode::STRING AS statuscode_label
  FROM <destination_database>.<destination_schema>.ACCOUNTS;

Dataverse column names are lowercase, so quote them in queries. The connector-managed columns are uppercase and don’t need quoting.

Manage table state¶

The connector maintains per-table ingestion state in the Dataverse Table State Service controller service. Each entry records the current ingestion status and the delta token used for change tracking.

View connector state¶

To view the current state of all tables:

1. Right-click on the canvas and select Controller services.
2. Locate the controller service named Dataverse Table State Service.
3. In the Dataverse Table State Service menu, click View state.

The state is a set of key/value pairs where the key is the table entity set name (for example, accounts). The value has the format <STATUS>;<deltaToken>;<skipToken>;<staleFlag>, for example:

accounts -> DONE;!AAAAAjE...;;

The STATUS can be one of the following:

• FETCHING — the connector is actively fetching records for this table.
• PROCESSING — the table is queued for ingestion but not currently being fetched.
• DONE — all available data was fetched successfully. The connector will check for new data on the next scheduled run according to the Ingestion Schedule Interval parameter.
• FAILED — an unrecoverable error occurred. Review the connector logs for details. If the logs indicate a configuration issue or a known limitation, resolve it and restart ingestion for the affected table. If no known cause is found, this may indicate a bug or an unsupported scenario; contact Snowflake Support.

Restart ingestion for a single table¶

Removing a table from the filter and re-adding it causes the connector to perform a full re-ingestion of that table. All currently available records are fetched again and merged into the destination table by primary key, so existing rows are updated in place rather than duplicated. However, the connector doesn’t recover delete operations that occurred during downtime.

To restart ingestion for a specific table:

1. Make sure the connector is running and open the connector process group.

2. Right-click on the canvas, select Parameters and remove the target table’s entity set name from the Source Tables Filter Value parameter.

3. Stop the List Dataverse Tables processor and start it again to trigger removal of the table state.

4. Wait until the table’s state has been removed:

   1. Right-click on the canvas and select Controller services.
   2. Locate the controller service named Dataverse Table State Service and select View state from its ellipsis menu.
   3. Wait until the table’s entity set name no longer appears in the list.

5. Stop the List Dataverse Tables processor and wait until all queues are empty.

6. (Optional) DROP the destination objects in Snowflake and let the connector recreate them.

   Since a full re-ingestion only upserts records that currently exist at the source, rows deleted at the source during downtime are not removed from the destination table automatically. After you delete the destination table and re-ingest it, the table includes only records that are still present in the source.

   DROP the following objects, where <TABLE> is the table’s entity set name in uppercase:

   • Destination table: <destination_database>.<destination_schema>.<TABLE>
   • Journal table: <destination_database>.<destination_schema>.<TABLE>__JOURNAL
   • Journal stream: <destination_database>.<destination_schema>.<TABLE>__JOURNAL_STREAM
   Copy code

   DROP STREAM IF EXISTS <destination_database>.<destination_schema>.<TABLE>__JOURNAL_STREAM;
   DROP TABLE IF EXISTS <destination_database>.<destination_schema>.<TABLE>__JOURNAL;
   DROP TABLE IF EXISTS <destination_database>.<destination_schema>.<TABLE>;
   

7. Re-add the table’s entity set name to Source Tables Filter Value and start the List Dataverse Tables processor again.

8. Verify that an entry for the table is present in Dataverse Table State Service state again.