Roman Klimenko
EN

Eksport af data fra CluedIn til Databricks eller Microsoft Fabric

cluedindatabricksmicrosoft-fabric

I denne artikel indlæser vi data fra CluedIn i en Databricks-notebook, udforsker og transformerer dem og gemmer resultatet i en Delta Lake-tabel.

CluedIn

Vores CluedIn-instans har 601.222 entiteter af typen /IMDb/Title.

Opret et API-token under Administration > API Tokens i CluedIn for at indlæse dem i Databricks:

CluedIn API token

Databricks

Installer afhængigheder

Installer cluedin-biblioteket for at forbinde til CluedIn API'et:

%pip install cluedin==2.2.0

Importer biblioteker

Importer de biblioteker, som notebooken bruger:

import pandas as pd
import matplotlib.pyplot as plt

import cluedin

Opret forbindelse til CluedIn

Angiv instansens URL og det API-token, vi oprettede tidligere, for at forbinde til CluedIn:

# CluedIn URL: https://foobar.mycluedin.com/:
#   - foobar is the organization's name
#   - mycluedin.com is the domain name

cluedin_context = {
  'domain': 'mycluedin.com',
  'org_name': 'foobar',
  'access_token': '(your token)'
}

Hent først én række for at undersøge dataene:

# Create a CluedIn context object.
ctx = cluedin.Context.from_dict(cluedin_context)

# GraphQL query to pull data from CluedIn.
query = """
query searchEntities($cursor: PagingCursor, $query: String, $pageSize: Int) {
  search(
    query: $query
    cursor: $cursor
    pageSize: $pageSize
    sort: FIELDS
    sortFields: {field: "id", direction: ASCENDING}
  ) {
    totalResults
    cursor
    entries {
      id
      name
      entityType
      properties
    }
  }
}
"""

# Fetch the first record from the `cluedin.gql.entries` generator.
next(cluedin.gql.entries(ctx, query, { 'query': 'entityType:/IMDb/Title', 'pageSize': 1 }))

Output:

{'id': '00001e32-9bae-53b9-a30f-cf30ed66c360',
 'name': 'Murder, Money and a Dog',
 'entityType': '/IMDb/Title',
 'properties': {'attribute-type': '/Metadata/KeyValue',
  'property-imdb.title.endYear': '\\N',
  'property-imdb.title.genres': 'Comedy,Drama,Thriller',
  'property-imdb.title.isAdult': '0',
  'property-imdb.title.originalTitle': 'Murder, Money and a Dog',
  'property-imdb.title.primaryTitle': 'Murder, Money and a Dog',
  'property-imdb.title.runtimeMinutes': '65',
  'property-imdb.title.startYear': '2010',
  'property-imdb.title.tconst': 'tt1664719',
  'property-imdb.title.titleType': 'movie'}}

Sortér GraphQL-resultaterne efter et unikt felt for at få stabil pagination og bedre performance. Entity ID fungerer godt:

    sort: FIELDS
    sortFields: {field: "id", direction: ASCENDING}

Indlæs nu hele datasættet i en Pandas DataFrame. Helperen nedenfor flader properties ud, fjerner unødvendige præfikser og erstatter punktummer med underscores, så navnene passer til Spark-skemaet:

ctx = cluedin.Context.from_dict(cluedin_context)

query = """
query searchEntities($cursor: PagingCursor, $query: String, $pageSize: Int) {
  search(
    query: $query
    sort: FIELDS
    cursor: $cursor
    pageSize: $pageSize
    sortFields: {field: "id", direction: ASCENDING}
  ) {
    totalResults
    cursor
    entries {
      id
      properties
    }
  }
}
"""

def flatten_properties(d):
    for k, v in d['properties'].items():
        if k == 'attribute-type':
            continue

        if k.startswith('property-'):
            k = k[9:] # len('property-') == 9

        k = k.replace('.', '_')

        d[k] = v

    del d['properties']

    return d

df_titles = pd.DataFrame(
    map(
        flatten_properties,
        cluedin.gql.entries(ctx, query, { 'query': 'entityType:/IMDb/Title', 'pageSize': 10_000 })))

df_titles.head()
no index

Sæt DataFramens indeks til entity ID:

df_titles.set_index('id', inplace=True)
df_titles.head()
index

Udforsk data

Lad os se, hvor mange film vi har per genre:

df_titles['imdb_title_genres'].str.split(',', expand=True).stack().value_counts().plot(kind='bar')
plt.title('Distribution of genres')
plt.xlabel('Genres')
plt.ylabel('Count')
plt.show()
genres

Opret skema

Opret nu et skema for dataene. Værdien imdb_title_genres er en string og ikke et array, så den skal splittes først:

from pyspark.sql import SparkSession
from pyspark.sql.types import StructType,StructField, StringType, ArrayType, IntegerType
from pyspark.sql.functions import split

spark = SparkSession.builder.getOrCreate()

schema = StructType([
    StructField('id', StringType(), True),
    StructField('imdb_title_endYear', StringType(), True),
    StructField('imdb_title_genres', ArrayType(StringType()), True),
    StructField('imdb_title_isAdult', StringType(), True),
    StructField('imdb_title_originalTitle', StringType(), True),
    StructField('imdb_title_primaryTitle', StringType(), True),
    StructField('imdb_title_runtimeMinutes', StringType(), True),
    StructField('imdb_title_startYear', StringType(), True),
    StructField('imdb_title_tconst', StringType(), True),
    StructField('imdb_title_titleType', StringType(), True)
    ])

df_spark_titles = spark.createDataFrame(df_titles)

df_spark_titles = df_spark_titles.withColumn('imdb_title_genres', split(df_spark_titles.imdb_title_genres, ','))

spark.sql('CREATE DATABASE IF NOT EXISTS cluedin')

df_spark_titles.write.mode('overwrite').format('parquet').saveAsTable('cluedin.imdb_titles', schema=schema)
display(df_spark_titles)

Dataene er nu tilgængelige i kataloget:

Databricks Catalog