SAP ERP 수집, 미디어와 BAPI 쓰기 반영

SAP ERP 커넥터는 온프레미스 ECC, 온프레미스 S/4HANA와 S/4HANA Cloud Private Edition에 연결합니다. SAP 애플리케이션 계층에 Palantir Foundry Connector 2.0 for SAP Applications 추가 기능을 설치해야 합니다.

애플리케이션 테이블, CDS 뷰, HANA Information View, 원격 BW Content Extractor, 트랜잭션 코드와 ABAP 보고서, 함수 모듈을 읽을 수 있습니다. SAP DMS(Document Management System)의 문서는 미디어 집합으로 수집할 수 있습니다.

소스 연결

Data Connection에서 + New Source를 선택하고 SAP ERP 소스를 선택합니다. 연결 방식은 다음 두 가지입니다.

Foundry가 SAP 호스트에 도달하도록 외부 통신 정책을 적용합니다. 온프레미스 시스템은 보통 에이전트 프록시 경로가 필요합니다. 새 소스에는 Foundry 워커를 사용하며 네트워크 경유지인 에이전트 프록시와 Legacy 에이전트 워커 실행을 구분합니다. 자체 서명 인증서로 SSL handshake가 실패하면 SAP의 올바른 인증서를 소스에 추가합니다.

SAP Data Accelerator

SAP Cloud Connector가 노출한 시스템에 SAP Data Accelerator의 관리형 연결 경로로 접근할 수 있습니다. Foundry 통합은 Beta이며 플랫폼 인스턴스별 제공 여부가 다를 수 있습니다. Foundry 워커가 필요합니다.

소스의 host와 port를 Cloud Connector의 가상 호스트와 포트로 지정하고 Palantir이 제공한 클라이언트 인증서를 적용합니다. Data Accelerator의 제어 영역과 데이터 영역으로 외부 통신을 허용하고 소스 속성 SAP Data Accelerator에 제어 영역 URL을 넣습니다.

배치 동기화

소스 개요에서 배치 동기화 옆의 + New를 선택합니다. 이름, 출력 데이터셋과 스케줄을 설정하고 전체 적재에는 SNAPSHOT, 증분 수집에는 APPEND를 선택합니다. SAP object type과 Object name을 지정합니다. 객체 이름을 입력하면 선택한 유형에 맞는 후보가 표시됩니다. 추가 설정은 Extras에서 구성합니다.

객체 유형별 연결 조건

증분 방식

• Multiple fields는 쉼표로 구분한 필드 중 어느 하나라도 이미 수집한 최댓값 이상인 행을 가져옵니다. 필드 조건은 OR로 결합합니다.
• Concatenate fields는 필드 값을 이어 붙인 값으로 증분 조건을 평가합니다.
• Change document table은 SAP 변경 문서를 사용하며 CDPOS only 또는 CDHDR and CDPOS를 선택합니다.
• Twin table은 별도 테이블의 필드가 증분 조건을 만족하는 경우 대상 테이블의 대응 행을 가져옵니다.
• SAP built-in delta는 extractor의 고유 변경 캡처 기능을 사용합니다.

증분 필드는 단조 증가하는 값이 적합합니다. 하루 도중 동기화하더라도 누락하지 않도록 비교가 >=이므로 경계값의 중복 행이 생길 수 있습니다. 후속 변환의 첫 단계에서 중복을 제거합니다. Transaction Code와 Function Module은 증분을 지원하지 않고 항상 스냅샷으로 실행합니다.

테이블형 증분 수집의 Max rows per sync는 실행당 대략적인 행 수를 제한합니다. 큰 테이블의 초기 수집을 여러 작은 실행으로 나누면 긴 네트워크 전송이 실패했을 때 다시 수행할 범위를 줄일 수 있습니다.

Twin table에서는 Incremental Twin Table에 보조 테이블을 지정하고 Incremental Twin Mapping에 다음 형태의 조인 조건을 넣습니다.

{PRIMARY_TABLE_NAME}-{FIELD_NAME}={TWIN_TABLE_NAME}-{FIELD_NAME}

여러 조건은 세미콜론으로 연결하며 AND로 평가합니다.

증분 상태 초기화

ERP Table, CDS View, HANA View와 BW Content Extractor에는 Reset incremental state를 사용할 수 있습니다. 활성화하면 SAP에서 전체 스냅샷을 읽어 데이터셋의 모든 파일을 교체하고 증분 상태를 다시 초기화합니다. 완료 후에는 토글을 꺼서 이후 실행이 증분 추가로 돌아가도록 합니다.

동기화 인자

행 필터와 동적 날짜

조건 그룹 안에 필드, 연산자와 값을 지정합니다. 필드 이름은 SAP 데이터 사전과 일치해야 합니다. 날짜 열의 고정 비교값은 YYYYMMDD 형식으로 입력합니다. 예를 들어 2021년 1월 1일은 20210101입니다. is like와 is not like는 * 와일드카드를 지원하며 A*12*는 A로 시작하고 12를 포함하는 문자열과 일치합니다.

추가 기능 SP26 이상에서는 필터 값에 동적 키워드와 날짜 계산 함수를 사용할 수 있습니다.

키워드를 함수에 넣거나 함수를 중첩할 수 있습니다. [ADDDAY([TODAY], 1)]은 다음 날을 계산하고 [GETDAY([ADDDAY([FIRSTDAYOFMONTH], -1)])]은 이번 달 첫날에서 하루를 뺀 날짜의 일을 구합니다. 키워드를 사용하면 날짜 문자열을 직접 조합할 필요가 없습니다. 날짜 인수를 직접 지정하는 예시는 [ADDDAY(22102022,1)]이며 결과는 23102022입니다.

Transaction Code

CDS View

WITH PARAMETERS로 선언한 뷰에는 CDS view parameters의 이름과 값 쌍을 전달합니다. 설정 자체는 선택 속성이지만 매개변수가 필요한 뷰에서는 값을 제공해야 합니다. 선언에 없는 이름이나 누락된 인자는 동기화를 실패시킵니다. 빈 행은 저장할 때 제거됩니다.

Function Module

Function input parameters에 이름과 값 쌍을 넣어 함수의 IMPORTING 인자로 전달합니다. 대부분의 BAPI에 필요합니다. Commit after function call은 기본적으로 꺼져 있습니다. 변경을 영속화할 때 커밋이 필요한 BAPI라면 활성화하여 호출 뒤 COMMIT WORK를 실행합니다. 아래 웹훅의 commit 기본값과 다릅니다.

고급 설정

일부 고급 설정은 소스 수준에서도 구성할 수 있습니다.

SAP DMS 문서 수집

소스 개요의 Media set syncs 옆에서 + New를 선택합니다. 미디어 집합 형식을 구체적으로 지정하면 해당 형식에 맞는 후속 변환을 사용할 수 있습니다. 원천 탐색에서 문서 유형별로 파일을 선택하거나 Filters를 직접 지정합니다.

Subfolder or media file에서 CPD 같은 Document Type을 선택하면 그 유형의 모든 문서를, 개별 문서를 선택하면 해당 파일 하나를 수집합니다.

문서 유형은 약어와 설명으로 표시합니다. 필터에는 약어를 사용할 수 있습니다.

웹훅으로 SAP 쓰기 반영

SAP 웹훅은 BAPI 같은 함수 모듈을 실행하여 원천 시스템을 변경합니다. Foundry 관리형 클라우드의 Foundry 워커 실행을 권장합니다. 이미 Foundry 워커인 소스는 별도 이관 없이 이 실행 경로를 사용하며 에이전트 워커 소스는 실행 방식을 전환해야 합니다.

작업 유형은 sap-erp-run-function-webhook-task-v0입니다. Task body는 Handlebars로 작성하고 입력값을 렌더링하면 JSON이 됩니다. 원격 시스템은 소스의 Remote (via Gateway)와 Context를 설정합니다.

단일 레코드

다음 템플릿의 입력 sales-doc-id는 판매 문서 번호, purchase-date는 YYYYMMDD 날짜입니다. json 헬퍼가 입력을 JSON 값으로 인코딩합니다. ORDER_HEADER_INX는 해당 필드를 갱신한다는 표시입니다.

{
    "function-name": "BAPI_SALESORDER_CHANGE",
    "inputs": {
        "SALESDOCUMENT": {{json sales-doc-id}},
        "ORDER_HEADER_IN": { "PURCH_DATE": {{json purchase-date}} },
        "ORDER_HEADER_INX": { "UPDATEFLAG": "U", "PURCH_DATE": "X" }
    },
    "output": "RETURN",
    "api-version": 2
}

여러 레코드와 병렬 실행

대량 및 병렬 쓰기 반영은 OAuth 2.0 인증을 지원하지 않습니다.

아래는 자재와 플랜트 번호를 가상 예시값으로 사용하여 구매 요청 두 건을 생성하는 본문입니다. 실제 실행하면 SAP에 쓰기가 발생하므로 대상 BAPI의 입력과 권한을 준비해야 합니다.

{
    "function-name": "BAPI_PR_CREATE",
    "commit": true,
    "parallel": { "max-jobs": 2, "max-calls-per-job": 4 },
    "inputs": [
        { "PRHEADER": { "PR_TYPE": "NB" }, "PRITEM": [ { "MATERIAL": "100-100", "PLANT": "1100", "QUANTITY": "1" } ] },
        { "PRHEADER": { "PR_TYPE": "NB" }, "PRITEM": [ { "MATERIAL": "100-200", "PLANT": "1100", "QUANTITY": "2" } ] }
    ],
    "output": ["NUMBER", "PRHEADEREXP"],
    "api-version": 2
}

inputs 배열에 객체가 두 개 이상 있거나 parallel 블록을 명시하면 SAP 측에서 병렬 실행합니다. parallel: {}처럼 빈 블록도 병렬 실행을 선택합니다. max-jobs는 동시에 실행할 작업 수, max-calls-per-job은 작업당 처리할 레코드 수입니다. 생략하면 SAP 추가 기능의 설정을 사용하며 기본값은 작업 3개, 작업당 호출 4개입니다. 레코드 수보다 작업 수를 크게 설정해도 남는 작업은 사용하지 않습니다.

응답과 업무 오류

다음 번호와 메시지는 응답 구조를 설명하는 가상 예시입니다.

단일 레코드이고 parallel이 없으며 출력 변수가 하나이면 그 변수의 행 목록을 반환합니다.

[
    { "NUMBER": "0010048439" }
]

단일 레코드에서 출력 변수 여러 개를 요청하면 변수 이름을 키로 하는 객체를 반환합니다. output["NUMBER"]처럼 접근합니다.

{
    "NUMBER": [ { "NUMBER": "0010048439" } ],
    "PRHEADEREXP": [ { "PREQ_NO": "0010048439", "PR_TYPE": "NB" } ]
}

병렬 쓰기 반영은 입력이 하나여도 입력 레코드별 원소를 가진 배열입니다.

[
    {
        "OUTPUT": { "NUMBER": [ { "NUMBER": "0010048439" } ], "PRHEADEREXP": [ { "PREQ_NO": "0010048439" } ] },
        "ROWNO": 1,
        "MSG": [ { "MSGTYP": "S", "MSGTXT": "Purchase requisition 0010048439 created" } ]
    }
]

OUTPUT은 해당 레코드의 결과이고 ROWNO는 1부터 시작하는 입력 배열의 위치입니다. MSG에는 레코드별 SAP 메시지가 들어갑니다. MSGTYP의 S는 성공, E는 오류, W는 경고, I는 정보, A는 중단이며 MSGTXT는 메시지 본문입니다.

HTTP 실패는 웹훅 오류로 반환합니다. HTTP 또는 웹훅 응답이 성공이어도 SAP 업무 오류가 포함될 수 있으므로 가능한 경우 RETURN을 요청하고 레코드별 MSG도 확인합니다.

추가 기능 버전과 사용자별 쓰기

여러 출력 변수, 대량 및 병렬 쓰기는 api-version: 2와 SAP 추가 기능 2.29.0 이상이 필요합니다. 버전을 생략한 기본 1은 단일 입력과 단일 출력 변수만 지원하며 parallel 블록을 사용하지 않습니다. 버전 2로 전환할 때에는 retries-per-request가 0보다 큰 설정을 제거합니다.

SAP Data Accelerator로 실제 사용자를 귀속시키는 쓰기 반영은 외부 인증 애플리케이션에 sap-erp-oauth2-token-webhook-task와 sap-erp-oauth2-refresh-webhook-task를 사용합니다. 애플리케이션과 사용자별 OAuth 권한을 구성하는 경로이며 앞의 대량/병렬 쓰기와 혼합하지 않습니다.

코드에서 SAP 추가 기능 API 사용

외부 변환이나 소스를 선언한 Python 함수에서 SAP 추가 기능 REST API를 호출할 수 있습니다. 기본 동기화와 웹훅으로 표현하기 어려운 호출 연쇄, Foundry 입력에 따른 문서 선택, 행별 BAPI 호출에 사용합니다. 다음 경로는 소스 URL의 /sap/palantir 뒤에 붙입니다.

변환에는 transforms-external-systems를 추가하고 SAP 소스를 가져옵니다. Foundry 입력을 같은 실행 작업에서 사용하려면 소스의 내보내기 제어에 허용할 조직과 보안 표지를 구성해야 합니다. Python 함수에는 소스의 코드 가져오기와 내보내기를 활성화하고 @function(sources=[...])로 선언합니다.

아래 변환들은 각각 독립 모듈입니다. 모듈 상수의 RID와 데이터셋 경로는 가상 예시이므로 실제 리소스로 수정합니다. 인증 정보는 가져온 소스의 HTTPS 클라이언트가 처리합니다. BAPI를 호출하기 전에 metadata로 대상 함수의 인자와 출력을 확인합니다.

행별 BAPI 호출 변환

입력 열 sales_doc_id와 purchase_date는 문자열이며 날짜는 YYYYMMDD 형식입니다. 호출마다 RETURN을 요청하고 HTTP 상태와 실제 SAP 응답을 기록합니다. 출력은 업무 성공만을 모은 목록이 아닙니다.

import json

import polars as pl
from transforms.api import Input, Output, transform
from transforms.external.systems import Source, external_systems

SOURCE_RID = "ri.magritte..source.00000000-0000-0000-0000-000000000001"
INPUT_PATH = "/Examples/SAP/sales-order-updates"
OUTPUT_PATH = "/Examples/SAP/writeback-results"


@external_systems(sap_source=Source(SOURCE_RID))
@transform.using(sales_order_updates=Input(INPUT_PATH), writeback_results=Output(OUTPUT_PATH))
def update_purchase_dates(sap_source, sales_order_updates, writeback_results):
    connection = sap_source.get_https_connection()
    base_url = connection.url.rstrip("/")
    client = connection.get_client()
    results = []
    for row in sales_order_updates.polars().iter_rows(named=True):
        response = client.post(
            base_url + "/v2/function/BAPI_SALESORDER_CHANGE/writeback",
            json={
                "commit": True,
                "input": {
                    "SALESDOCUMENT": row["sales_doc_id"],
                    "ORDER_HEADER_IN": {"PURCH_DATE": row["purchase_date"]},
                    "ORDER_HEADER_INX": {"UPDATEFLAG": "U", "PURCH_DATE": "X"},
                },
                "output": ["RETURN"],
            },
        )
        response.raise_for_status()
        results.append({
            "sales_doc_id": row["sales_doc_id"],
            "purchase_date": row["purchase_date"],
            "status_code": response.status_code,
            "sap_return": json.dumps(response.json(), ensure_ascii=False),
        })
    writeback_results.write_table(pl.DataFrame(results, schema={
        "sales_doc_id": pl.String,
        "purchase_date": pl.String,
        "status_code": pl.Int32,
        "sap_return": pl.String,
    }))

commit: true는 커밋이 있어야 변경을 저장하는 표준 BAPI에 필요합니다. 중간 HTTP 오류가 나면 앞서 반영된 SAP 변경을 Foundry가 롤백하지 않습니다.

동기 BAPI 호출 함수

리포지토리에 SAP 소스를 sap_source 이름으로 가져온 뒤 함수에 선언합니다. Workshop 액션이나 다른 함수 호출자가 판매 문서 번호와 날짜를 전달합니다.

import json
from functions.api import function
from functions.sources import get_source


@function(sources=["sap_source"])
def update_sales_order_purchase_date(sales_doc_id: str, purchase_date: str) -> str:
    source = get_source("sap_source")
    connection = source.get_https_connection()
    client = connection.get_client()
    response = client.post(
        connection.url.rstrip("/") + "/v2/function/BAPI_SALESORDER_CHANGE/writeback",
        json={
            "commit": True,
            "input": {
                "SALESDOCUMENT": sales_doc_id,
                "ORDER_HEADER_IN": {"PURCH_DATE": purchase_date},
                "ORDER_HEADER_INX": {"UPDATEFLAG": "U", "PURCH_DATE": "X"},
            },
            "output": ["RETURN"],
        },
        timeout=30,
    )
    response.raise_for_status()
    return json.dumps(response.json(), ensure_ascii=False)

반환값은 실제 SAP RETURN 구조를 포함한 JSON 문자열입니다. 호출자는 SAP의 오류와 경고를 해석하여 사용자에게 보여 줘야 하며 HTTP 2xx를 업무 성공으로 대체하지 않습니다.

병렬 BAPI 호출 변환

아래 REST 본문의 필드명은 웹훅 Task body와 다릅니다. 웹훅의 inputs 대신 bulkinput, max-jobs 대신 paralleljobs, max-calls-per-job 대신 functioncallperjob을 사용합니다.

import json

import polars as pl
from transforms.api import Input, Output, transform
from transforms.external.systems import Source, external_systems

SOURCE_RID = "ri.magritte..source.00000000-0000-0000-0000-000000000001"
INPUT_PATH = "/Examples/SAP/sales-order-updates"
OUTPUT_PATH = "/Examples/SAP/parallel-writeback-results"


@external_systems(sap_source=Source(SOURCE_RID))
@transform.using(sales_order_updates=Input(INPUT_PATH), writeback_results=Output(OUTPUT_PATH))
def update_purchase_dates_parallel(sap_source, sales_order_updates, writeback_results):
    connection = sap_source.get_https_connection()
    client = connection.get_client()
    bulkinput = [
        {
            "SALESDOCUMENT": row["sales_doc_id"],
            "ORDER_HEADER_IN": {"PURCH_DATE": row["purchase_date"]},
            "ORDER_HEADER_INX": {"UPDATEFLAG": "U", "PURCH_DATE": "X"},
        }
        for row in sales_order_updates.polars().iter_rows(named=True)
    ]
    results = []
    if bulkinput:
        response = client.post(
            connection.url.rstrip("/") + "/v2/function/BAPI_SALESORDER_CHANGE/parallel_writeback",
            json={
                "commit": True,
                "output": ["RETURN"],
                "bulkinput": bulkinput,
                "parallel": True,
                "paralleljobs": 5,
                "functioncallperjob": 10,
            },
        )
        response.raise_for_status()
        results = [
            {"index": i, "result": json.dumps(result, ensure_ascii=False)}
            for i, result in enumerate(response.json())
        ]
    writeback_results.write_table(pl.DataFrame(results, schema={"index": pl.Int64, "result": pl.String}))

paralleljobs는 SAP 백그라운드 작업 수, functioncallperjob은 작업당 호출 수입니다. SAP의 가용 처리량에 맞게 조정합니다. parallel: false이면 한 번의 대량 요청 안에서 순차 처리합니다. 예제의 index는 응답 배열을 0부터 열거한 번호이고 SAP의 1 기반 ROWNO와 다릅니다. SAP가 반환한 원래 필드는 result JSON에 보존합니다. 입력이 비어 있으면 호출 없이 명시적 스키마의 빈 결과를 기록합니다.

DMS 파일을 미디어 집합으로 가져오기

소스와 출력 미디어 집합을 가져오고 실제 리소스 RID로 변경합니다. 아래 문서 유형 CPT와 형식 pdf는 예시 필터입니다. 목록 응답에서 DOCUMENT_ID와 DOCUMENT_NAME을 읽고 ID로 파일을 다운로드합니다.

import io
from transforms.api import transform
from transforms.external.systems import Source, external_systems
from transforms.mediasets import MediaSetOutput

SOURCE_RID = "ri.magritte..source.00000000-0000-0000-0000-000000000001"
MEDIA_SET_RID = "ri.mio.main.media-set.00000000-0000-0000-0000-000000000002"


@external_systems(sap_source=Source(SOURCE_RID))
@transform(documents=MediaSetOutput(MEDIA_SET_RID))
def fetch_dms_documents(sap_source, documents):
    connection = sap_source.get_https_connection()
    base_url = connection.url.rstrip("/")
    client = connection.get_client()
    list_response = client.get(
        base_url + "/v2/get_documents",
        params={"DOCUMENTTYPE": "CPT", "FILEFORMAT": "pdf"},
    )
    list_response.raise_for_status()
    for document in list_response.json():
        response = client.get(
            base_url + "/v2/get_document",
            params={"DOCUMENT_ID": document["DOCUMENT_ID"]},
        )
        response.raise_for_status()
        documents.put_media_item(io.BytesIO(response.content), document["DOCUMENT_NAME"])

Foundry 입력의 구매 주문 EBELN 목록에 해당하는 문서만 읽으려면 /v2/get_documents의 OBJECTLINK 필터를 사용하거나 입력의 DOCUMENT_ID를 /v2/get_document에 전달합니다. 원천 DMS와 Foundry 데이터셋을 함께 사용하는 경우에도 소스의 내보내기 제어가 필요합니다.

STXL의 긴 텍스트는 압축 해제를 거쳐 수집하며, BW Content Extractor와 HANA 외부 뷰는 SAP 측 추출 및 게시 설정을 먼저 준비합니다. 사용자별 쓰기 반영과 커스텀 권한은 SAP 추가 기능의 역할 및 인증 구성을 함께 관리해야 합니다.