Connect to Workday and query data from Snowflake

Connect to Workday and query data from Snowflake (Workday 연결 및 Snowflake에서 데이터 쿼리)

이 주제는 Snowflake Notebook을 만들고, Workday Python 커넥터를 설치하며, 자격 증명을 구성하고, Workday 데이터에 대해 쿼리를 실행하는 방법을 설명해요.

출처: Snowflake Documentation

본문

Preview 기능 — 열림

모든 계정에서 사용할 수 있어요.

Note

Snowflake용 Workday Live Data Query는 Workday에서는 Early Adopter(EA)이고 Snowflake에서는 Preview 상태예요. 접근을 요청하려면 Workday 계정 담당자에게 연락해요.

이 주제를 시작하기 전에 Snowflake for Workday Live Data Query 설정을 완료해요.

Step 1: Notebook 생성

Snowflake Workspace에서 새 Python notebook을 만들어요. 모든 LDQ 연결·쿼리 코드가 여기서 실행돼요.

  1. Snowsight에서 Workspaces로 이동해요.
  2. 기존 workspace를 열거나 + Workspace를 클릭해 새로 만들어요.
  3. workspace 안에서 + > Notebook을 클릭해요.

Step 2: 외부 접근 통합 연결

Notebook은 Workday에 아웃바운드 호출을 하려면 명시적 권한이 필요해요. Snowflake for Workday Live Data Query 설정에서 만든 WORKDAY_LDQ_TEST_EAI 통합을 연결해요.

  1. notebook을 열어요.
  2. notebook 상단의 Connected 옆 드롭다운 화살표를 클릭해요.
  3. 서비스 세부 정보 패널에서 **Enabled External Access Integrations (EAIs)**를 확인해요. WORKDAY_LDQ_TEST_EAI가 나열되어 있지 않으면 Manage service를 클릭해 추가해요. 거기에도 보이지 않으면 + Create new service를 클릭해 새 서비스를 만들고 EAI를 연결해요.
  4. 메시지가 표시되면 notebook 세션을 다시 시작해요.

Important

EAI가 연결되지 않으면 Workday로의 모든 아웃바운드 HTTP 호출이 네트워크 오류로 실패해요.

Step 3: Python 커넥터 설치

스테이지에서 workday_ldq 패키지를 notebook의 런타임 환경에 설치해요. 이렇게 하면 DataServiceConfig와 create_connection API를 사용할 수 있어요.

설치는 세션 단위이며 notebook을 재시작할 때마다 다시 실행해야 해요. 커넥터의 종속성(예: trino, requests, lz4)은 번들되지 않으므로 pip가 PyPI에서 자동으로 다운로드하며, 그래서 네트워크 규칙에 pypi.org와 files.pythonhosted.org가 포함돼요.

notebook의 첫 번째 셀에서 실행해요:

import sys
import subprocess

from snowflake.snowpark.context import get_active_session

session = get_active_session()

# Replace with the correct wheel filename
session.file.get(
    "@WORKDAY_LDQ_TEST.LIVEDATA.LDQ_STAGE/ldq_python_client-1.0.3-py3-none-any.whl",
    "/tmp"
)

subprocess.check_call([
    sys.executable, "-m", "pip", "install",
    "/tmp/ldq_python_client-1.0.3-py3-none-any.whl",
    "--quiet"
])

print("Wheel installed successfully!")

Important

wheel은 notebook 재시작 간 지속되지 않아요. notebook을 재시작하거나 다시 연결할 때마다 이 셀을 다시 실행해요. 항상 먼저 실행되도록 첫 번째 셀로 유지해요.

Step 4: 자격 증명 구성

커넥터는 DataServiceConfig()를 통해 메모리 내 사전 형태로 자격 증명을 받아요. 개인 키는 임시 UDF를 사용해 런타임에 Snowflake Secret에서 검색돼요. 키는 디스크에 절대 기록되지 않으며 notebook에 지속되지 않아요.

notebook 셀에서 실행해요:

from snowflake.snowpark.context import get_active_session
from snowflake.snowpark.functions import udf
from workday_ldq import DataServiceConfig

session = get_active_session()
session.sql("USE DATABASE WORKDAY_LDQ_TEST").collect()
session.sql("USE SCHEMA LIVEDATA").collect()

@udf(
    name="get_secret_temp",
    is_permanent=False,
    replace=True,
    external_access_integrations=["WORKDAY_LDQ_TEST_EAI"],
    secrets={"pk": "WORKDAY_LDQ_TEST.LIVEDATA.WORKDAY_PRIVATE_KEY"}
)
def get_secret_temp() -> str:
    import _snowflake
    return _snowflake.get_generic_secret_string("pk")

private_key_pem = session.sql("SELECT get_secret_temp()").collect()[0][0]

config = DataServiceConfig({
    "wd.authn.clientId":            "<your_client_id>",
    "wd.authn.isu":                 "<your_isu_username>",
    "wd.authn.accessTokenEndpoint": "https://<host>/ccx/oauth2/<tenant>/token",
    "wd.authn.privateKey":          private_key_pem,
    "wd.host":                      "<your_workday_host>",
    "wd.port":                      "443"
})

del private_key_pem
session.sql("DROP FUNCTION IF EXISTS get_secret_temp()").collect()
print("Config created successfully!")

자리 표시자 값을 Workday 관리자가 제공한 값으로 바꿔요.

속성 설명
wd.authn.clientId Workday Register API Client 작업의 Client ID.
wd.authn.isu ISU 사용자 이름(예: snowflake_ldq_user).
wd.authn.accessTokenEndpoint https:///ccx/oauth2//token 형식의 전체 OAuth2 토큰 URL.
wd.authn.privateKey 런타임에 Snowflake Secret에서 안전하게 검색되는 PEM 개인 키 내용.
wd.host Workday 서비스 호스트.
wd.port 항상 443.

Note

Workspace notebook은 _snowflake나 st.secrets를 통해 Snowflake Secrets에 직접 접근할 수 없어요. 임시 UDF는 Snowflake 실행 환경 내부에서(_snowflake를 사용할 수 있는 곳) 실행되어 시크릿을 검색하고 notebook 세션으로 반환해요. UDF는 사용 후 즉시 드롭되고, 개인 키는 del로 메모리에서 지워져요. 키는 디스크에 절대 기록되지 않아요.

Note

Snowflake 환경이 프록시를 통해 트래픽을 라우팅한다면 wd.host를 <proxy_host>로 설정하고 wd.authn.accessTokenEndpoint를 프록시의 토큰 엔드포인트를 가리키도록 업데이트해요.

Step 5: 연결 및 쿼리 실행

Workday LDQ 서비스에 연결

config 객체로 연결을 열어요. 성공적인 연결은 인증, 네트워킹, 구성이 모두 올바르다는 것을 확인해 줘요.

from workday_ldq import create_connection

connection = create_connection(config)
print("Connected successfully!")

쿼리 실행

커서를 사용해 Workday 데이터 서비스에 SQL을 보내요. 쿼리는 Snowflake 테이블이 아닌 Workday의 Unified Data Catalog에 대해 실행돼요.

cursor = connection.cursor()
cursor.execute("SELECT COUNT(*) FROM workday_core.public.worker")
results = cursor.fetchall()
print(results)
cursor.close()

결과를 DataFrame으로 로드

import pandas as pd

cursor = connection.cursor()
cursor.execute("SELECT * FROM workday_core.public.worker LIMIT 100")

columns = [desc[0] for desc in cursor.description]
rows    = cursor.fetchall()
df      = pd.DataFrame(rows, columns=columns)
cursor.close()

df.head()

연결 닫기

작업이 끝나면 항상 연결을 닫아 Snowflake와 Workday 양쪽의 리소스를 해제해요.

connection.close()

샘플 쿼리

Note

이 쿼리들은 Workday 데이터를 대상으로 하며 Snowflake 테이블이 아니므로 LDQ 커넥터를 통해 실행해야 해요. 각 쿼리 문자열을 Snowflake Worksheet에 직접 붙여넣지 말고 위에 보인 대로 cursor.execute() 호출에 붙여넣어요.

아래 예는 카탈로그와 스키마로 workday_core.public을 사용해요. 환경은 다른 이름을 사용할 수 있어요. 항상 먼저 발견 쿼리를 실행해 테넌트에서 무엇을 사용할 수 있는지 확인해요.

사용 가능한 카탈로그와 스키마 발견

SHOW CATALOGS

SHOW SCHEMAS IN workday_core

SHOW TABLES IN workday_core.public

모든 근로자 수

SELECT COUNT(*) AS total_workers
FROM workday_core.public.worker

직함이 있는 활성 근로자 나열

SELECT
    w.worker_id,
    w.full_name,
    w.employee_type,
    jp.job_title
FROM workday_core.public.worker w
JOIN workday_core.public.job_profile jp
    ON w.job_profile_id = jp.job_profile_id
WHERE w.active = TRUE
LIMIT 50

관리 수준별 인원

SELECT
    management_level,
    COUNT(*) AS headcount
FROM workday_core.public.worker
WHERE active = TRUE
GROUP BY management_level
ORDER BY headcount DESC

최근 90일 동안 채용된 근로자

SELECT
    worker_id,
    full_name,
    hire_date,
    business_title
FROM workday_core.public.worker
WHERE hire_date >= CURRENT_DATE - INTERVAL '90' DAY
ORDER BY hire_date DESC

Note

EA에서 사용 가능한 객체는 Workforce와 Talent로 제한돼요. 행 수준 보안 제어는 GA에서 계획돼요. 테이블·컬럼 수준 접근은 Workday의 ISU 보안 그룹에 의해 제어돼요.

다음 단계

데이터가 DataFrame에 있으면 Cortex Code를 사용해 쿼리를 작성하고, 시각화를 구축하고, Workday 데이터에 대한 AI 지원 분석을 받을 수 있어요.

더 알아보기 (Learn more)