스트리밍과 배칭을 통합한 실시간 피쳐 엔지니어링 플랫폼 구축 실습
- PHPz앞으로
- 2023-04-04 11:35:011517검색
이 글은 주로 기능 엔지니어링 개발 분야에서 Alibaba Cloud FeatHub 프로젝트 팀의 플랫폼 실습과 구축 경험을 공유합니다.
이 공유는 네 부분으로 나누어져 있습니다. 첫 번째 부분은 일반적으로 FeatHub가 기능 개발, 배포, 모니터링 및 공유 프로세스에서 직면한 시나리오, 목표, 문제점 및 과제를 소개합니다. 두 번째 부분은 FeatHub의 아키텍처 사고 방식을 소개합니다. . 및 관련 핵심 개념, 세 번째 부분에서는 FeatHub 사용 중 API의 기본 사용법, 기본 컴퓨팅 기능 및 샘플 시나리오의 코드 실습뿐만 아니라 성능 최적화, 향후 확장 목표 및 오픈 소스의 공동 구축을 소개합니다. 프로젝트 정보를 제공하는 소스 커뮤니티 학습, 개발 및 사용을 위해 FeatHub의 기록 데이터 재생 기능, 오프라인, 니어라인, 온라인 처리 지원, Alibaba Cloud의 업스트림 및 다운스트림 구성 요소에 대한 지원도 공유할 것입니다.
1. FeatHub가 필요한 이유
1. 대상 시나리오
(1) Python 환경이 필요한 데이터 과학자
현재 인기 있는 머신러닝 추론 및 훈련 프로그램의 대부분은 기본적으로 Python Written을 사용하여 데이터 과학자가 작성합니다. , 일부 전통적인 기계 학습 시나리오에서 사용되는 인기 있는 TensorFlow, PyTorch 및 scikit-learn 등이 있습니다. 우리는 데이터 과학자가 익숙한 Python을 계속 사용하여 기능 엔지니어링 코드를 작성하여 엔드투엔드 기계 학습 링크의 개발 및 배포를 완료하고 익숙한 Python 생태계의 라이브러리를 사용할 수 있도록 지원하기를 희망합니다.
(2) 실시간 기능 생성
점점 더 많은 기계 학습 응용 프로그램이 실시간 방향으로 개발되고 있으며, 실시간 처리를 통해 기계 학습의 효율성과 정확성을 향상시킬 수 있습니다. 목표를 달성하려면 실시간 기능을 생성해야 합니다. 이는 단순히 쿼리 특성을 실시간으로 얻는 것이 아니라 실시간으로 특성을 생성하는 것이기도 합니다. 예를 들어, 지난 2분 동안의 사용자 클릭 수를 실시간으로 얻으려면 스트리밍 컴퓨팅 엔진을 사용하여 실시간 기능 계산을 완료해야 합니다.
(3) 멀티 클라우드 배포를 지원하려면 오픈 소스 솔루션이 필요합니다
생산 보안을 보장하고 클라우드 공급업체 사이에서 입찰 이점을 얻기 위해 멀티 클라우드 배포를 달성하려는 중소기업이 점점 더 많아지고 있습니다. 따라서 우리 솔루션은 사용자가 클라우드 공급업체에 바인딩할 것을 요구하지 않고 사용자가 다양한 클라우드 공급업체 중에서 자유롭게 선택할 수 있도록 하며 프라이빗 클라우드에 기능 엔지니어링 작업을 배포할 수도 있습니다.
이것은 FeatHub 프로젝트가 처음 설립되었을 때 충족되기를 바랐던 조건 중 일부입니다.
2. 실시간 기능 엔지니어링의 문제점
오늘날 많은 회사에서 실시간 기능 엔지니어링 작업을 개발하고 있습니다. 개발, 배포, 모니터링 및 후속 공유를 포함하여 기능의 전체 수명 주기를 포괄하는 몇 가지 문제점이 있습니다.
(1) 개발이 어렵다
① 기능 탐색
개발 단계에서는 실시간 기능 프레임워크인 Apache Flink를 주로 사용하는데, Flink는 기본적으로 실시간 스트림 컴퓨팅의 사실상의 표준이기 때문이며, 하지만 Flink를 사용하거나 유사한 프레임워크를 사용하여 실시간 기능을 개발할 때 기능 탐색의 어려움을 해결할 필요가 있습니다. 많은 데이터 사이언티스트들이 특성 순회를 해결한 경험이 없으며, 이러한 문제를 해결하려면 많은 학습 시간과 비용이 필요하며, 이는 개발 단계의 주요 문제점입니다.
(2) 배포가 어렵습니다
1 수동 번역이 필요합니다
많은 회사에는 데이터 과학자가 작성한 단일 프로세스 Python 작업을 분산 방식으로 실행할 수 있는 Flink 또는 Spark 작업으로 변환하는 전담 플랫폼 팀이 있습니다. 고성능 달성 고가용성 배포. 번역 프로세스는 전체 개발 수명주기의 길이를 증가시킵니다. 그리고 번역 작업을 하려면 추가적인 인력이 필요하기 때문에 개발 비용이 증가하고 버그가 유입될 가능성이 더욱 높아집니다. 다른 그룹의 사람들이 데이터 과학자의 작업을 번역한 후의 논리는 원래 논리와 일치하지 않을 수 있으며, 이로 인해 더 많은 디버깅 작업이 발생하게 됩니다.
(3) 모니터링 난이도 높음
① 기능 분포 변경
특성 엔지니어링 작업의 전체 품질과 효율성은 작업에 버그가 있는지 여부에 따라 달라질 뿐만 아니라 훈련 중 데이터 수치 분포에 근접하는 등 특정 특성을 충족하는 업스트림 입력 데이터 수치 분포에 따라 달라집니다. 업스트림 작업에서 생성된 데이터 분포의 변화로 인해 많은 작업의 추론 성능이 저하되는 경우가 많습니다. 이 경우 개발자는 전체 링크를 세그먼트별로 추적하여 기능 데이터 분포가 변경된 위치를 확인하고 특정 상황에 따라 재교육이나 버그 수정이 필요한지 확인해야 합니다. 이 부분 인력의 과도한 업무량도 문제점이다.
(4) 공유가 어렵습니다
1 개발 작업의 중복
많은 특성 계산 작업의 개발 팀과 시나리오가 다르지만 실제로는 유사하거나 심지어 동일한 특성 정의가 사용됩니다. 많은 회사에는 회사 내의 여러 팀이 기존 기능을 쿼리하고 재사용할 수 있는 좋은 채널이 없습니다. 이로 인해 여러 팀에서 반복적인 개발을 수행해야 하는 경우가 많으며 동일한 기능에 대한 일부 기능을 생성하기 위해 작업을 반복적으로 실행해야 하는 경우도 있습니다. 동일한 기능을 생성하려면 더 많은 컴퓨팅, 메모리 및 저장 공간이 필요하기 때문에 인력과 컴퓨팅/스토리지 리소스가 낭비됩니다.
② 특정 시점의 올바른 의미
모든 사람이 특성 교차가 무엇인지 이해할 수 있도록 위 그림은 이 문제를 보여주는 간단한 예를 제공합니다. 그림 왼쪽 상단의 표는 사용자의 행동 특성으로, 특정 ID를 가진 사용자가 서로 다른 시간 노드에서 지난 2분간 클릭한 횟수를 나타냅니다. 이 클릭수는 사용자가 광고를 클릭할지 여부를 추론하는 데 도움이 될 수 있습니다. 훈련에 이러한 기능을 사용하려면 일반적으로 레이블을 사용하여 기능을 일부 사용자 데이터 세트에 연결해야 합니다. 그림의 왼쪽 아래 테이블은 사용자가 실제로 광고를 클릭했는지 여부에 대한 일부 긍정적인 샘플과 부정적인 샘플의 데이터 세트를 보여주며, 사용자가 서로 다른 시점에 생성한 긍정적인 샘플 또는 부정적인 샘플을 표시합니다. 이 두 데이터 세트의 기능을 연결하여 훈련 데이터 세트를 형성하려면 일반적으로 사용자 ID를 키로 기반으로 기능을 연결해야 합니다. 타임스탬프를 고려하지 않고 단순히 Table Join을 수행할 경우 Feature Crossing 문제가 발생할 수 있습니다. 예를 들어 6시 3분에 사용자가 최근 2분간 클릭한 횟수는 10이어야 하는데, 스플라이싱을 통해 구한 특징값은 7시부터 6이 될 수 있다. 이러한 종류의 특징 교차는 실제 추론 효과를 감소시킵니다. 특정 시점의 올바른 의미를 갖는 조인 결과는 아래 그림과 같아야 합니다.
샘플을 접합할 때 특징 교차를 방지하려면 위 그림의 왼쪽 테이블에 있는 각 데이터 조각에 대해 여러 버전 특성 중 왼쪽 테이블의 타임스탬프보다 작고 가장 가까운 타임스탬프를 갖는 특성 값을 찾아서 최종 생성된 학습 데이터 세트에 연결합니다. 특정 시점의 올바른 의미론과의 연결은 위 그림의 오른쪽에 표시된 훈련 데이터 세트를 생성합니다. 다양한 시점에 대해 지난 2분 동안 생성된 해당 특성 값이 있습니다. 이렇게 생성된 학습 데이터 세트는 학습 및 추론 성능을 향상시킬 수 있습니다. 3. Feature Store의 핵심 시나리오
다음으로 Feature Store로서의 FeatHub를 소개하고, FeatHub가 해결하려는 문제와 전체 기능 개발 주기의 각 단계에서 제공하는 도구를 소개합니다. 
(3) 기능 알람
기능 모니터링 단계에서 개발자가 기능 값 분포의 변화를 신속하게 감지하고 대응할 수 있도록 FeatHub는 향후 기능과 같은 일반적인 기능 품질 문제를 다루는 몇 가지 공통 지표를 생성할 것입니다. 이러한 지표를 기반으로 잘못된 값 또는 기능 평균 및 알람이 발행되어 담당자에게 즉시 통보되어 관련 기능 분포의 변경 원인을 조사하고 대응하여 엔드 투 엔드 추천 링크의 효과를 유지합니다.
(4) 기능 공유
기능 공유 단계에서 FeatHub는 향후 기능 등록 및 검색 기능을 제공하여 같은 회사 내 여러 팀의 개발자가 원하는 기능이 이미 존재하는지 쿼리하고 재사용할 수 있도록 합니다. 기능 정의 그리고 생성된 특징 데이터.
위 그림은 FeatHub의 핵심 기능을 보여줍니다. 개발 단계에서 FeatHub는 기능 접합, 기능 집계 및 특정 시점의 올바른 의미 체계를 갖춘 기타 로직을 지원하는 사용하기 쉬운 SDK를 제공할 수 있습니다. 배포 단계에서 FeatHub는 높은 처리량, 낮은 대기 시간 기능 생성을 지원하고 Flink를 기능 계산을 위한 실행 엔진으로 사용하도록 지원하며 여러 기능 스토리지 시스템을 지원할 수 있어 사용자가 사용하려는 스토리지 유형을 자유롭게 선택할 수 있습니다. 모니터링 단계에서 FeatHub는 오프라인 및 실시간 모니터링을 포함하여 기능 분포의 변화를 모니터링하는 실시간 지표를 제공하여 개발자가 적시에 문제를 찾을 수 있도록 지원합니다. 공유 단계에서 FeatHub는 개발자가 기능을 등록, 검색 및 재사용할 수 있도록 사용하기 쉬운 웹 UI 및 SDK를 제공합니다.
피처스토어 분야에는 올해 초 링크드인에서 오픈소스화한 Feathr, 수년간 오픈소스화해온 Feast 등 대표적인 피처스토어 프로젝트가 있습니다. 우리는 이러한 프로젝트를 조사한 결과 우리가 제안한 목표 시나리오를 달성하는 데 적합하지 않다는 것을 발견했습니다.
기존 솔루션과 비교하여 FeatHub는 다음과 같은 추가 가치를 제공합니다.
① 간단하고 사용하기 쉬운 Python SDK. FeatHub의 SDK는 기존 Feature Store 프로젝트의 SDK를 참조하며 이러한 프로젝트의 핵심 기능을 지원하고 SDK의 추상화 기능과 사용 편의성을 더욱 향상시킵니다.
② 단일 머신에서 개발과 실험을 지원합니다. 개발자는 실험을 실행하기 위해 분산된 Flink 또는 Spark 클러스터에 연결할 필요가 없으며 단일 시스템의 CPU 또는 메모리 리소스만 사용하여 개발 및 실험을 수행할 수 있으며 다음과 같은 단일 시스템에서 기계 학습 알고리즘 라이브러리를 사용할 수 있습니다. scikit-learn.
③ 코드 수정 없이 실행 엔진을 전환할 수 있습니다. 사용자가 단일 머신에서 개발을 완료한 후, 기능 계산 로직을 표현하는 코드를 수정하지 않고도 단일 머신 실행 엔진을 Flink 또는 Spark와 같은 분산 실행 엔진으로 전환할 수 있습니다. Flink를 실행 엔진으로 사용하면 Feathhub는 높은 처리량, 낮은 지연 시간의 실시간 기능 계산을 지원할 수 있습니다. FeatHub는 향후 Spark를 실행 엔진으로 사용하는 것을 추가로 지원하여 사용자가 오프라인 시나리오에서 잠재적으로 더 나은 처리량 성능을 얻고 시나리오에 따라 가장 적합한 실행 엔진을 자유롭게 선택할 수 있도록 할 것입니다.
④ 실행 엔진의 확장 기능을 제공합니다. FeatHub는 Flink 및 Spark를 실행 엔진으로 지원할 뿐만 아니라 개발자가 실행 엔진을 맞춤화하고 기능 ETL을 위해 회사 내부에서 개발한 실행 엔진을 사용할 수 있도록 지원합니다.
⑤ 코드는 오픈 소스이므로 사용자는 FeatHub를 배포할 클라우드 공급업체를 자유롭게 선택하거나 프라이빗 클라우드에 배포할 수 있습니다.
2. FeatHub 아키텍처 및 핵심 개념
1. Architecture
위는 FeatHub의 주요 모듈을 포함하는 아키텍처 다이어그램입니다. 최상위 계층은 사용자 정의 데이터 소스, 데이터 끝점 및 기능 계산 논리를 지원하기 위한 Python SDK 세트를 제공합니다. SDK에서 정의한 기능은 기능 메타데이터 센터에 등록할 수 있으므로 다른 사용자와 작업이 기능을 쿼리하고 재사용할 수 있으며 기능 메타데이터를 기반으로 기능 계보를 추가로 분석할 수도 있습니다. 기능 정의에는 기능의 소스 및 싱크뿐만 아니라 UDF 호출, 기능 접합, 오버 윈도우 및 슬라이딩 윈도우 기반 집계 등과 같은 일반적인 계산 논리도 포함됩니다. 사용자 정의 기능을 생성해야 하는 경우 FeatHub는 기존 기능의 계산 논리를 실행하기 위해 일부 내장 기능 프로세서, 즉 실행 엔진을 제공합니다. 사용자가 단일 컴퓨터에서 실험을 수행해야 하는 경우 로컬 프로세서를 사용하여 원격 클러스터에 연결할 필요 없이 단일 컴퓨터의 리소스를 사용할 수 있습니다. 실시간 기능을 생성해야 하는 경우 Flink 프로세서를 사용하여 높은 처리량, 낮은 지연 시간의 스트리밍 기능 계산을 완료할 수 있습니다.
향후에는 온라인 특성 계산을 구현하기 위해 Lambda 함수와 유사한 기능 서비스도 지원될 수 있으며, Spark에 연결하여 처리량이 높은 오프라인 특성 계산을 완료할 수 있습니다. 실행 엔진은 온라인 기능 저장소에 Redis를 사용하고, 오프라인 기능 저장소에 HDFS를 사용하고, Near-line 기능 저장소에 Kafka를 사용하는 등 다양한 오프라인 및 온라인 기능 저장소 시스템과 인터페이스할 수 있습니다.
위 그림은 사용자가 FeatHub를 사용하고 다운스트림 기계 학습 교육 및 추론 프로그램과 연결하는 방법을 보여줍니다. 사용자 또는 개발자는 SDK를 통해 계산하려는 기능을 표현한 다음 실행 엔진에 제출합니다. 전개. 특성이 계산된 후에는 Redis 및 HDFS와 같은 특성 저장소로 출력되어야 합니다. 기계 학습 오프라인 교육 프로그램은 일괄 교육을 위해 HDFS의 데이터를 직접 읽을 수 있습니다. 온라인 기계 학습 추론 프로그램은 온라인 추론을 위해 Redis의 데이터를 직접 읽을 수 있습니다.
2. 핵심 개념
위 그림은 FeatHub의 핵심 개념 간의 관계를 보여줍니다. TableDescriptor는 기능 모음을 나타냅니다. TableDescriptor는 논리적 변환을 통해 새로운 TableDescriptor를 생성할 수 있습니다.
TableDescriptor는 두 가지 범주로 나뉩니다. FeatureTable은 특정 물리적 주소를 가진 테이블을 표현합니다. 예를 들어 Redis의 테이블일 수도 있고 HDFS의 테이블일 수도 있습니다. FeatureView는 물리적 주소가 반드시 필요하지 않은 논리적 테이블입니다. 일반적으로 일련의 논리적 문자열 변환 후 FeatureTable에서 가져옵니다.
FeatureView에는 다음과 같은 3개의 하위 클래스가 있습니다.
① DerivedFeatureView 출력 특성 테이블과 해당 입력 특성 테이블(예: 소스)의 행은 기본적으로 일대일입니다. 단일 라인 변환 논리(예: 더하기, 빼기, 곱하기 및 나누기), 창 집계 논리 및 기능 접합 논리의 표현을 지원할 수 있습니다. 훈련 데이터를 생성하는 데 사용할 수 있습니다. 예를 들어, 이전에 소개된 예에서 실제 훈련 데이터를 얻기 위해 다른 차원 테이블의 기능과 훈련 샘플을 연결해야 하는 경우 DerivedFeatureView를 사용하여 이를 완료할 수 있습니다.
② SlidingFeatureView는 슬라이딩 창으로 계산된 특징 표현을 지원합니다. 출력 특성 테이블과 입력 특성 테이블의 행이 반드시 일대일일 필요는 없습니다. 슬라이딩 윈도우로 계산된 특성값은 새로운 입력이 없더라도 시간이 지나면서 변하기 때문이다. SlidingFeatureView는 실시간으로 생성된 특성을 유지하고 온라인 추론을 위해 Redis와 같은 온라인 특성 저장소에 출력하는 데 사용할 수 있습니다. 예를 들어 SlidingFeatureView를 사용하여 지난 2분 동안 각 사용자가 특정 웹 페이지를 클릭한 횟수를 계산하고 기능 값을 Redis에 실시간으로 업데이트한 다음 광고 추천 링크를 통해 이 기능의 값을 쿼리할 수 있습니다. 온라인 추론을 위한 온라인.
3 OnDemandFeatureView 는 Feature Service와 함께 사용하여 온라인 기능 계산을 지원할 수 있습니다. 예를 들어 Amap을 사용할 때 개발자는 사용자의 요청을 받은 후 사용자의 현재 물리적 위치와 마지막 요청이 전송된 물리적 위치를 기반으로 사용자의 이동 속도와 방향을 계산하여 경로 추천을 지원할 수 있습니다. 결정. 이러한 기능은 사용자 요청을 받으면 온라인으로 계산되어야 합니다. OnDemandFeatureView는 이러한 시나리오를 지원하는 데 사용될 수 있습니다.
Transform은 특성 계산 논리를 표현합니다. FeatHub는 현재 다음 5가지 유형의 특징 계산 로직을 지원합니다.
① Expression을 사용하면 사용자가 DSL 언어를 기반으로 한 줄의 특징 계산 로직을 표현할 수 있습니다. 표현 능력은 SQL 언어의 select 문에 가깝고, 내장 함수 호출은 물론 덧셈, 뺄셈, 곱셈, 나눗셈도 지원해 SQL에 익숙한 개발자도 빠르게 시작할 수 있다.
② Join은 기능 접합 논리를 표현합니다. 개발자는 차원 테이블의 이름, 접합할 기능의 이름과 같은 정보를 지정할 수 있습니다.
3 PythonUDF 기능을 계산하는 사용자 정의 Python 함수를 지원합니다.
4 OverWindow 는 Over Window 집계 로직을 표현합니다. 예를 들어, 데이터 행을 수신할 때 사용자는 이전 5개 데이터 행을 집계하고 특정 규칙과 일치하는 데이터 조각 수를 계산하려고 합니다.
⑤ SlidingWindow는 슬라이딩 윈도우 집계 로직을 표현합니다.
위 그림에서 볼 수 있듯이 일반적으로 기능 ETL 작업은 기능 소스 테이블에서 기능을 읽고, 여러 기능 계산 논리를 통해 새 기능을 생성하고, 생성된 기능을 기능 결과 테이블에 출력합니다. 기능 소스 테이블은 FileSystem, Kafka, Hive 등과 같은 다양한 기능 저장소에 연결할 수 있습니다. 마찬가지로 기능 결과 테이블은 FileSystem, Kafka, Redis와 같은 기능 저장소에 연결할 수도 있습니다.
Processor에는 독립형 물리적 리소스, 분산 Flink 클러스터 및 분산 Spark 클러스터를 각각 사용하여 사용자 정의 기능 계산 논리를 실행할 수 있는 LocalProcessor, FlinkProcessor 및 SparkProcessor가 포함됩니다.
3. FeatHub API 표시
1. 기능 계산 기능
FeatHub의 아키텍처와 핵심 개념을 소개한 후 몇 가지 샘플 프로그램을 사용하여 FeatHub SDK의 표현력과 사용 편의성을 보여드리겠습니다. 기능 개발 SDK의 핵심 기능은 새로운 기능 계산 로직을 표현하는 방법입니다. FeatHub SDK는 기능 접합, 창 집계, 내장 함수 호출 및 사용자 정의 Python 기능을 지원하며 향후 JAVA 또는 C++ 기반 UDF 호출도 지원할 예정입니다.
위 그림은 기능 접합의 코드 조각을 보여줍니다. 이 예에서는 HDFS에 사용자의 구매 행동을 기록하는 원본 긍정 및 부정 샘플 데이터가 있다고 가정합니다. 우리는 사용자가 각 제품을 구매할 때 제품 가격을 추가로 얻고 싶습니다. Price_updates 테이블은 제품 가격 변동에 대한 데이터를 유지합니다. 제품 가격이 변경될 때마다 제품 ID 및 최신 제품 가격을 포함하여 가격_업데이트 테이블에 데이터 행이 생성됩니다. JoinTransform을 사용하고 table_name=price_updates, feature_name=price 및 key=item_id를 설정하여 해당 기능 접합 논리를 표현할 수 있습니다. 이러한 방식으로 FeatHub는 Price_updates에서 주어진 item_id가 있는 행을 찾고 타임스탬프를 기반으로 가장 적절한 가격 값을 찾아 샘플 데이터 테이블에 결합할 수 있습니다.
Over Window 집계 코드 조각은 OverWindowTransform을 사용하여 기능을 계산하는 방법을 보여줍니다. 사용자는 expr=”item_counts * 가격” 및 agg_fun=”SUM”을 사용하여 구매한 품목의 수량과 가격을 기준으로 최근 기간의 총 소비량을 계산할 수 있습니다. 창 길이는 2분입니다. group_by_keys=["user_id"]는 각 사용자에 대해 개별적으로 해당 총 소비량을 계산한다는 의미입니다.
슬라이딩 윈도우 집계는 오버 윈도우 집계와 유사하지만 API의 유일한 차이점은 step_size를 추가로 지정할 수 있다는 것입니다. step_size=1분이면 창이 슬라이드되어 1분마다 새로운 특성 값을 생성합니다.
내장 함수 호출용 코드 조각은 DSL 언어를 사용하여 덧셈, 뺄셈, 곱셈, 나눗셈 및 UDF 호출을 표현하는 방법을 보여줍니다. 입력 데이터에 택시 승하차의 타임스탬프가 포함되어 있다고 가정합니다. UNIX_TIMESTAMP 내장 함수를 호출하여 승객 승하차 타임스탬프를 정수형의 에포크 시간으로 변환한 후, 획득한 에포크 시간을 빼서 각 여정의 길이를 구할 수 있는 기능으로 활용 가능 후속 훈련 및 추론을 위해.
PythonUDF가 호출하는 코드 조각에서 사용자는 Python 함수를 사용자 정의하여 소문자 문자열 생성과 같은 입력 기능에 대한 임의 처리를 수행할 수 있습니다.
위의 코드 조각을 통해 FeatHub의 API가 비교적 간단하고 사용하기 쉽다는 것을 알 수 있습니다. 사용자는 처리 엔진의 세부 사항을 알지 못한 채 계산 논리에 필요한 매개변수만 설정하면 됩니다.
2. 샘플 시나리오
위의 샘플 시나리오에서 사용자에게는 두 개의 데이터 소스가 있습니다. 구매 이벤트에는 Kafka 또는 FileSystem에서 제공될 수 있는 사용자가 구매한 제품의 샘플 데이터가 포함되어 있으며, 항목 가격 이벤트에는 제품 가격 변동에 대한 데이터가 포함되어 있습니다. 품목 가격이 변경될 때마다 품목 ID 및 최신 품목 가격을 포함하여 품목 가격 이벤트에 데이터 행이 생성됩니다. 제품을 구매하는 사용자의 각 샘플 데이터에 대해 해당 행동이 발생한 마지막 2분간의 사용자 총 소비량을 계산하고, 이를 사용자가 특정 제품을 구매할 것인지 유추하는 데 도움이 되는 기능으로 활용할 수 있기를 바랍니다. . 이 기능을 생성하려면 위 그림에 설명된 계산 논리를 사용하여 먼저 item_id를 Join_key로 사용하여 품목 가격 이벤트의 가격 기능을 구매 이벤트에 연결할 수 있습니다. 그런 다음 기간을 기준으로 집계하고 user_id를 group_by _keys로 사용하여 지난 2분 동안 각 사용자의 총 소비량을 계산합니다.
3. 샘플 코드
위의 코드 조각은 샘플 FeatHub 애플리케이션을 위해 완료해야 하는 단계를 보여줍니다.
① 먼저 사용자는 FeatHubClient를 생성하고 processor_type을 설정해야 합니다. 로컬 실험인 경우 로컬로 설정할 수 있으며, 원격 분산 프로덕션 배포인 경우 Flink로 설정할 수 있습니다.
② 사용자는 데이터를 읽으려면 소스를 생성해야 합니다. 예를 들어 FileSystemSource를 사용하여 오프라인 스토리지 시스템에서 데이터를 읽거나 KafkaSource를 사용하여 니어라인 스토리지 시스템에서 실시간 데이터를 읽을 수 있습니다. FileSystemSource에서 사용자는 data_format, 스키마, 파일 위치 등과 같은 정보를 지정할 수 있습니다. 사용자는 데이터 소스 테이블에서 시간을 나타내는 열과 해당 파싱 형식을 각각 표현하기 위해 time_stamp_field와 time_stamp_format을 제공할 수 있다는 점에 주목할 필요가 있습니다. FeatHub는 이 정보를 사용하여 특정 시점의 올바른 기능 계산을 완료하여 기능 교차 문제를 방지합니다.
3 사용자는 FeatureView를 생성하여 기능 결합 및 집계 논리를 표현할 수 있습니다. 접합을 원하는 경우 사용자는 item_price_events.price를 사용하여 접합하려는 기능을 표현할 수 있습니다. FeatHub는 item_price_events라는 테이블을 찾아 거기에서 가격이라는 기능을 가져옵니다. 사용자는 OverWindowTransform을 사용하여 기간 초과 집계를 완료하고 total_pay_last_two_ Minutes라는 특성을 정의할 수도 있습니다. 여기서 window_size=2분은 지정된 표현식과 집계 함수를 적용하여 2분 이내에 데이터의 특징을 계산한다는 의미입니다.
4 정의된 FeatureView에 대해 사용자가 로컬에서 개발 및 실험하고 단일 머신에서의 학습을 위해 scikit-learn 알고리즘 라이브러리를 사용하려는 경우 to_pandas() API를 사용하여 데이터를 가져올 수 있습니다. Pandas DataFrame 형식의 단일 머신.
⑤ 사용자가 기능의 프로덕션 배포를 완료해야 하는 경우 FileSystemSink를 사용하여 데이터 저장을 위한 오프라인 기능 저장소를 지정할 수 있습니다. 그런 다음 Execute_insert()를 호출하여 지정된 싱크에 기능을 출력합니다.
FeatHub의 기본 가치는 사용자가 기능을 쉽게 개발할 수 있는 SDK와 기능을 계산할 수 있는 실행 엔진을 제공하는 것입니다. 또한 FeatHub는 실행 엔진의 성능 최적화도 제공하여 사용자가 기능 배포 단계에서 더 많은 이점을 얻을 수 있도록 합니다. 예를 들어 슬라이딩 윈도우 집계 기반 기능의 경우 현재 기본 Flink API를 사용하여 계산하는 경우 Flink는 기능 값 변경 여부에 관계없이 각 슬라이딩 step_size에서 해당 기능 값을 출력합니다. window_size=1시간, step_size=1초인 슬라이딩 윈도우의 경우 Flink는 대부분의 경우 동일한 기능 값을 출력할 수 있습니다. 이는 네트워크 트래픽, 다운스트림 스토리지 및 기타 리소스를 낭비하게 됩니다. FeatHub는 사용자가 슬라이딩 창의 동작을 구성할 수 있도록 지원하여 기능 계산 작업의 리소스 사용을 최적화하기 위해 기능 값이 변경될 때 슬라이딩 창이 기능만 출력하도록 허용합니다.
또한 FeatHub는 슬라이딩 창의 메모리 및 CPU 사용량을 더욱 최적화합니다. 일부 시나리오에서는 사용자가 유사한 여러 슬라이딩 창 기능을 사용하게 됩니다. 이러한 기능은 창 크기만 다릅니다. 예를 들어, 지난 1분, 5분, 10분 동안 각 사용자가 구매에 지출한 총 금액을 확인하려고 할 수 있습니다. 기본 Flink API가 계산에 사용되는 경우 작업은 세 가지 집계 연산자를 사용하여 이러한 세 가지 기능을 각각 계산할 수 있습니다. 각 집계 연산자에는 별도의 메모리 공간이 있습니다. 이러한 연산자가 처리하는 데이터와 계산 논리가 겹치는 부분이 크다는 점을 고려하면 FeatHub는 사용자 지정 연산자를 사용하여 이러한 기능의 계산을 균일하게 완료하여 메모리 및 CPU 리소스 절약이라는 목표를 달성할 수 있습니다.
FeatHub는 현재 GitHub의 오픈 소스이며 일부 기본 LocalProcessor 및 FlinkProcessor 기능을 지원할 수 있습니다. 우리는 사용자 기능 엔지니어링의 개발과 구현을 촉진하기 위해 FeatHub의 핵심 기능을 더욱 개선할 것입니다. 여기에는 보다 일반적으로 사용되는 오프라인 스토리지 및 온라인 스토리지 지원, 노트북과의 도킹, 기능 메타데이터를 시각화하는 웹 UI 제공, 사용자의 기능 등록, 검색 및 재사용 지원, FeatHub의 실행 엔진으로 Spark 사용 지원 등이 포함됩니다.
FeatHub 코드 베이스: https://github.com/alibaba/FeatHub
FeatHub 코드 예제: https://github.com/flink-extended/FeatHub-examples
FeatHub 코드 베이스는 현재 github/alibaba 디렉토리에 있습니다. 모든 사람이 FeatHub 사용 방법을 더 쉽게 배우고 필요한 시나리오의 요구 사항을 충족하는 코드 조각을 빠르게 찾아 참조할 수 있도록 flink-extended/feathub-examples 코드 라이브러리에 추가 코드 예제를 제공합니다. 자유롭게 사용하고 사용해 보세요. 누구나 피드백을 제공하고 PR에 기여할 수 있습니다.
4. Q&A 세션
Q1: point_in_time 조인 중에 데이터 순서가 뒤바뀌어 지연되거나 Join을 수동으로 작성하여 기능 교차가 발생합니까?
A1: 원칙적으로 데이터의 순서가 맞지 않더라도 가입 시 타임스탬프 필드를 고려하지 않으면 순서가 맞지 않을 수 있습니다. 실제 시나리오에서는 원본 데이터의 순서가 잘못될 수도 있습니다. 이때 Flink와 유사한 워터마크 전략을 사용하여 늦게 도착하는 데이터를 기다리며 순서가 뒤바뀌는 영향을 줄일 수 있습니다. 또한 일반 오프라인 작업을 사용하여 온라인 기능 데이터를 다시 채울 수 있으므로 데이터 장애의 영향을 더욱 줄일 수 있습니다.
Q2: FeatHub가 출시된 후 과거 훈련 데이터와 해당 기능을 어떻게 생성하나요? FeatHub는 기록 데이터 재생을 지원합니까?
A2: FeatHub API는 재생을 지원할 수 있지만 이 기능 부분은 아직 생산 검증되지 않았습니다. FeatHub는 Flink 및 Spark를 실행 엔진으로 사용하도록 지원하므로 Flink 및 Spark의 컴퓨팅 기능을 재사용하여 기록 데이터 재생을 완료할 수 있습니다. 예를 들어 Spark 작업을 시작하고 지난달 HDFS의 모든 데이터를 처리하도록 소스를 설정하고 정의된 기능 접합 및 집계 논리를 실행한 다음 계산된 기능을 출력할 수 있습니다.
Q3. FeatHub는 오프라인 기능 계산만 담당하는데, 일부 온라인 기능은 어떻게 처리하나요?
A3: 특성 계산은 오프라인, 니어라인, 온라인으로 구분됩니다 Flink는 지난 5분 동안 사용자 클릭 수와 같은 특성을 실시간으로 계산할 수 있는 니어라인 실행 엔진입니다. 오프라인 계산을 지원합니다. 따라서 FeatHub는 오프라인 및 니어라인 기능 계산을 지원할 수 있습니다. FeatHub는 향후 OnDemandFeatureView에서 표현되는 특징을 계산하기 위해 Feature Service 기반 아키텍처를 사용하여 온라인 특징 계산을 지원할 계획입니다.
Q4: FeatHub는 Alibaba Cloud에서 서비스를 제공합니다. 현재 ODPS 등 업스트림 및 다운스트림 생태학적 지원은 무엇입니까?
A4: FeatHub는 ODPS, Holo 및 Alibaba Cloud에서 제공하는 기타 서비스를 포함하여 Flink에서 지원하는 모든 소스/싱크를 지원합니다. 현재 FeatHub는 Kafka 및 FileSystem만 지원합니다. 점차적으로 더 많은 스토리지 지원을 추가할 예정입니다.
위 내용은 스트리밍과 배칭을 통합한 실시간 피쳐 엔지니어링 플랫폼 구축 실습의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!