NLP Data Factory 1.x¶

This section is for those familiar with NLP Data Factory 1.x (that is, any version of NLP Data Factory prior to 2.0). There are some key differences between the current version of NLP Data Factory and these earlier versions. This section explains the motivation behind the changes, covers the key differences, and guides you through migrating from NLP Data Factory 1.x to the current version.

Note

This section is full of historical context and migration advice. If you’re starting with NLP Data Factory 2.0 or later, you can safely ignore this section.

Motivation¶

The Airflow-based NLP Data Factory offers lots of benefits compared to NLP Data Factory 1.x:

Airflow has a range of useful features that aren’t present in NLP Data Factory 1.x. These include several UI views for visualizing workflows, fine-grained access control, scheduling workflows, and more.
Airflow is extremely flexible. Tasks can be chained together in any number of ways, and custom functionality is easy to add to a workflow. For example, you can build a workflow that runs a query in I2E, uses a machine learning model, reads a file from Amazon S3, runs a few lines of your own Python code to perform some post-processing, and sends an email notification.
Airflow can be extended to work with a multitude of external services, using third-party providers. Airflow’s provider index lists the providers maintained by the Airflow community, but there are even more providers out there.

Key differences¶

NLP Data Factory 1.x	NLP Data Factory 2.0 and later
Custom-built orchestrator; doesn’t use Airflow.	Provider for Airflow; a popular open-source orchestrator.
Distributed as RPMs or Docker images.	The IQVIA NLP Airflow provider is distributed as a Python wheel. Airflow itself has several installation methods, including Docker images.
Workflows are created via the Data Factory API.	DAGs are defined as Python files.
Workflows have fixed behaviors, but custom behavior can be added with webhooks.	DAGs are flexible by nature, and easy to extend with custom behavior. A huge number of third-party providers are available for integrating Airflow with external services. Existing NLP Data Factory webhooks can still be used in DAGs.
Limited ability to chain stages together.	Easy to define complex dependencies and orderings between tasks.
No support for scheduling workflows.	Support for scheduling DAGs.
Documents are uploaded to the Data Factory API during run submission.	DAGs pull documents from an external location (e.g. Amazon S3 or an SQL database).
Copies of documents and I2E results are stored in NLP Data Factory’s PostgreSQL database.	No copies of documents or I2E results are stored in an intermediate database.
NLP Data Factory Connectors can read from external datastores, and write I2E results to external datastores.	Transfer operators are provided (e.g. `SQLTimestampToI2EOperator`, `I2EToS3Operator`) to read data from external datastores, and write I2E results to external datastores.
One I2E server per NLP Data Factory agent. Additional agents can be added to scale up the system.	The number of Airflow workers is unrelated to the number of I2E servers. Additional Airflow workers or I2E servers can be added to scale up the system.
I2E resource bundles are distributed across NLP Data Factory agents by uploading the bundle to the NLP Data Factory API, pausing the agents, and resuming the agents.	I2E resource bundles are distributed to I2E servers using the `LocalFilesystemToI2EResourceBundleOperator` or `S3ToI2EResourceBundleOperator`.
Limited access control. There are two types of NLP Data Factory users: admin and non-admin.	Airflow has fine-grained Role-Based Access Control (RBAC). Custom roles can be created with specific permissions.
Supports OpenID Connect, SAML, LDAP, and Remote User authentication.	Supports OAuth, LDAP, and Remote User authentication.
Secrets and credentials in configuration files can be stored in Azure Key Vault.	Secrets and credentials used by DAGs can be stored in a variety of external secret management services, such as Azure Key Vault or AWS Secrets Manager.
The `data-factory-manager health` and `data-factory-agent-manager health` CLI commands perform a variety of health checks.	Airflow provides several CLI commands and HTTP endpoints for Checking Airflow Health Status.

Migration¶

Migrating environments¶

This section covers the steps to migrate from an NLP Data Factory 1.x environment to an equivalent Airflow environment. Refer to the Installation section for more details.

Create an Airflow environment (if you don’t have one already).
Install the IQVIA NLP Airflow provider in your Airflow environment.
Provision I2E servers for Airflow to use. You could re-purpose existing I2E servers associated with NLP Data Factory agents. Ensure the I2E servers are accessible from your Airflow workers. Create an I2E connection in Airflow for each I2E server.
If your workflows require I2E resource bundles, then you need to write an Airflow DAG to distribute these bundles to your I2E servers. This DAG could be run manually, or it could be scheduled to automatically deploy new I2E resource bundles. The i2e_resource_bundles.py example DAG demonstrates how to do this.
If your workflows interact with an external datastore (such as SQL or S3), then ensure the datastore is accessible from your Airflow workers, and create an Airflow connection for the datastore.

Migrating workflows¶

Each workflow in NLP Data Factory 1.x must be converted to an Airflow DAG. You can look through our Example DAGs to find some DAGs that match what you want to do, but we’ll do an example of the conversion process here.

Let’s take the following query-document workflow as an example. We want to convert this workflow into a DAG:

{
  "name": "sdoh_workflow",
  "max_retries": 3,
  "i2e_index_template_path": "data_factory/sdoh-2_0_0/sdoh",
  "i2e_index_template_overrides": {
    "useDaemon": true,
    "checkDaemon": true,
    "ontologyIndexSuppression": "All"
  },
  "i2e_query_templates": [
    {
      "template_path": "data_factory/sdoh-2_1_0/All Social Determinants of Health",
      "result_options": {
        "query_results": true,
        "highlight": false,
        "cached_documents": false,
        "highlighted_documents": "none"
      }
    }
  ],
  "i2e_query_template_overrides": {
    "checkClasses": false
  }
}

Firstly, we need to upload the source data to I2E. In NLP Data Factory 1.x, this would be done by uploading the source data to the NLP Data Factory server. In the current version of NLP Data Factory, transfer operators are used to copy source data from an external datastore to I2E. For this example, let’s use the S3ToI2EOperator to copy a source data file from an S3 bucket into I2E:

from iqvia_nlp_provider.transfers.s3 import S3ToI2EOperator

i2e_source_data_uri = S3ToI2EOperator(
    aws_conn_id="aws_conn_id",
    i2e_conn_id="i2e_default",
    aws_bucket_name="s3-bucket",
    aws_key="source.txt",
)

Next, we need to build an index in I2E. To do this, we use the I2EMakeIndexOperator. Compare the fields here with the i2e_index_template_path and i2e_index_template_overrides fields from our workflow:

from iqvia_nlp_provider.operators.i2e import I2EMakeIndexOperator

i2e_index_uri = I2EMakeIndexOperator(
    i2e_conn_id="i2e_default",
    i2e_source_data_uri=i2e_source_data_uri.output,
    i2e_index_template="data_factory/sdoh-2_0_0/sdoh",
    i2e_index_settings_overrides={
        "useDaemon": True,
        "checkDaemon": True,
        "ontologyIndexSuppression": "All",
    },
)

Next, we need to run our query on this index. To do this, we use the I2ERunQueryOperator. Again, compare the fields here with the i2e_query_template_path and i2e_query_template_overrides fields from our workflow:

from iqvia_nlp_provider.operators.i2e import I2ERunQueryOperator

i2e_query_results_uri = I2ERunQueryOperator(
    i2e_conn_id="i2e_default",
    i2e_index_uri=i2e_index_uri.output,
    i2e_query_template="data_factory/sdoh-2_1_0/All Social Determinants of Health",
    i2e_query_settings_overrides={
        "checkClasses": False,
    },
)

Now we need to retrieve the query results. In NLP Data Factory 1.x, the query results would be stored in NLP Data Factory’s PostgreSQL database and could then be retrieved from the NLP Data Factory API. In the current version of NLP Data Factory, transfer operators are used to copy query results from I2E to an external datastore. For this example, let’s use the I2EToS3Operator to copy the query results from I2E to an S3 bucket:

from iqvia_nlp_provider.transfers.s3 import I2EToS3Operator

I2EToS3Operator(
    i2e_conn_id="i2e_default",
    aws_conn_id="aws_conn_id",
    i2e_query_results_uri=i2e_query_results_uri.output,
    aws_bucket_name="s3-bucket",
)

The last step is to delete any temporary resources we’ve created in I2E. Now that our query results have been copied to S3, there’s no need to keep the source data, index or query results in I2E. We use the I2EDeleteResourcesOperator to delete these resources from I2E:

from iqvia_nlp_provider.operators.i2e import I2EDeleteResourcesOperator

 I2EDeleteResourcesOperator(
    i2e_conn_id="i2e_default",
    i2e_resource_uris=[i2e_source_data_uri.output, i2e_index_uri.output, i2e_query_results_uri.output],
)

Finally, let’s put all these operators together in a single DAG, with an appropriate name and the same number of retries as our workflow. We use the chain function to ensure our tasks run in the correct order.

from airflow.sdk import dag
from airflow.sdk.bases.operator import chain
from iqvia_nlp_provider.operators.i2e import I2EDeleteResourcesOperator, I2EMakeIndexOperator, I2ERunQueryOperator
from iqvia_nlp_provider.transfers.s3 import I2EToS3Operator, S3ToI2EOperator

@dag(schedule=None, catchup=False, default_args={"retries": 3})
def i2e_query_sdoh():

    s3_to_i2e_task = i2e_source_data_uri = S3ToI2EOperator(
        aws_conn_id="aws_conn_id",
        i2e_conn_id="i2e_default",
        aws_bucket_name="s3-bucket",
        aws_key="source.txt",
    )

    i2e_make_index_task = i2e_index_uri = I2EMakeIndexOperator(
        i2e_conn_id="i2e_default",
        i2e_source_data_uri=i2e_source_data_uri.output,
        i2e_index_template="data_factory/sdoh-2_0_0/sdoh",
        i2e_index_settings_overrides={
            "useDaemon": True,
            "checkDaemon": True,
            "ontologyIndexSuppression": "All",
        },
    )

    i2e_run_query_task = i2e_query_results_uri = I2ERunQueryOperator(
        i2e_conn_id="i2e_default",
        i2e_index_uri=i2e_index_uri.output,
        i2e_query_template="data_factory/sdoh-2_1_0/All Social Determinants of Health",
        i2e_query_settings_overrides={
            "checkClasses": False,
        },
    )

    i2e_to_s3_task = I2EToS3Operator(
        i2e_conn_id="i2e_default",
        aws_conn_id="aws_conn_id",
        i2e_query_results_uri=i2e_query_results_uri.output,
        aws_bucket_name="s3-bucket",
    )

    i2e_delete_resources_task = I2EDeleteResourcesOperator(
        i2e_conn_id="i2e_default",
        i2e_resource_uris=[i2e_source_data_uri.output, i2e_index_uri.output, i2e_query_results_uri.output],
    )

    chain(
        s3_to_i2e_task,
        i2e_make_index_task,
        i2e_run_query_task,
        i2e_to_s3_task,
        i2e_delete_resources_task,
    )


i2e_query_sdoh()

Migrating webhooks¶

If your query-document workflow uses webhooks, you can continue to use these webhooks in Airflow. Our transfer operators accept legacy_before_hook and legacy_after_hook arguments that can be used to call your existing webhooks during DAG runs.

Firstly, you will need to create an Airflow connection for your webhook server. To create a new connection from the Airflow web interface, select the Admin tab, open the Connections page, and create a new connection of type Legacy NLP Data Factory Webhook. This connection type has the following fields:

Connection Id	The connection ID. This must be unique across the Airflow installation.
Connection Type	The connection type. This should be set to `Legacy NLP Data Factory Webhook`.
Description	A human-readable description for the connection.
Host	The hostname of the webhook server. This shouldn’t include the schema or port.
Schema	The schema of the connection, e.g. `http` or `https`. Defaults to `http`.
Login	Optional username to authenticate with the webhook server.
Password	Optional password to authenticate with the webhook server.
Port	The port number of the webhook server. Defaults to `80` for `http` and `443` for `https`.
Extras	Extra fields can be provided as a JSON object here. The available extra fields are `timeout`, `proxy`, `stream`, `verify_ssl`, `cert`, and `allow_redirects`.

Once the connection has been created, you can pass the connection ID to transfer operators in your DAGs to call your webhook server. For example, if you created a connection with the ID webhook_default, then you would pass it to the S3ToI2EOperator with the legacy_before_hook argument like so:

from iqvia_nlp_provider.transfers import Webhook
from iqvia_nlp_provider.transfers.s3 import S3ToI2EOperator

S3ToI2EOperator(
    ...,
    legacy_before_hook=Webhook(
        conn_id="webhook_default",
        endpoint="before-hook/",
        # This argument determines which result files will be posted in the webhook call.
        # This example shows the default values. See the Webhook and ResultOptions class
        # documentation for more details.
        result_options=ResultOptions(
            query_results=True,
            cached_documents=True,
            highlight=True,
            highlighted_documents="content_and_stylesheet",
        ),
    )
)

This would send the source data to the before-hook/ endpoint on your webhook server, before uploading the source data to I2E.

Similarly, you can use the I2EToS3Operator with the legacy_after_hook and i2e_source_data_uri arguments like so:

from iqvia_nlp_provider.transfers import Webhook
from iqvia_nlp_provider.transfers.s3 import I2EToS3Operator

I2EToS3Operator(
    ...,
    legacy_after_hook=Webhook(
        conn_id="webhook_default",
        endpoint="after-hook/",
    ),
    i2e_source_data_uri=i2e_source_data_uri.output,
)

This would send the I2E query results to the after-hook/ endpoint on your webhook server, before uploading the query results to S3.

Note that if the webhook step fails, the entire task associated with the operator will fail as well. This is unlike the legacy Data Factory, where a run whose webhook step failed would return a SUCCESS_WITH_WARNINGS status. You can simulate this in Airflow by changing the transfer operator task’s settings to not fail the entire DAG run if it fails.

Migrating client code¶

You should update any client code making requests to the NLP Data Factory API. The client code should be reworked to interact with the Airflow 3 API server and the Airflow REST API instead.

The following example shows how to trigger a DAG run using Airflow’s REST API:

from requests import post

token = post(
    url="https://airflow_api_server/auth/token",
    json={"username": "airflow_username", "password": "airflow_password"},
).json()["access_token"]

post(
    url="https://airflow_api_server/api/v2/dags/<dag_id>/dagRuns",
    json={},
    headers={"Authorization": f"Bearer {token}"},
)

token=$(
    curl --request POST \
         --url 'https://airflow_api_server/auth/token' \
         --json '{"username":"airflow_username","password":"airflow_password"}' \
    | jq -r '.access_token'
)

curl --request POST \
     --url 'https://airflow_api_server/api/v2/dags/<dag_id>/dagRuns' \
     --json '{}' \
     --header "Authorization: Bearer ${token}"

The following example shows how to retrieve the status of a DAG run using Airflow’s REST API:

from requests import get

response = get(
    url="https://airflow_api_server/api/v2/dags/<dag_id>/dagRuns/<dag_run_id>",
    headers={"Authorization": f"Bearer {token}"},
)
state = response.json()["state"]
print(f"Status: {state}")

response=$(
    curl --request GET \
         --url 'https://airflow_api_server/api/v2/dags/<dag_id>/dagRuns/<dag_run_id>' \
         --header "Authorization: Bearer ${token}"
)
state=$(echo "${response}" | jq -r '.state')
echo "Status: ${state}"

Alternatively, your DAGs can be designed to remove the need for client code. For example, if your DAGs are scheduled to run periodically, and they pull documents from an external location and write I2E results to another external location, then this could remove the need for client code that submit runs or retrieves results.

Migrating authentication¶

You can see a comparison of available authentication options in Data Factory 1.x and Airflow in the below table:

External Authentication method	NLP Data Factory 1.x	Apache Airflow
LDAP	✔️	✔️
OpenID	❌	✔️
OpenID Connect	✔️	❌
SAML	✔️	❌
Remote User	✔️	✔️*
OAuth	❌	✔️

* Some custom code is required to use Remote User authentication in Airflow.

For more information on configuring authentication in Airflow, see the Authentication section.