Zenginleştirme Boru Hatları

Veri Zenginleştirme İş Akışları Oluşturun

Süreç madenciliği veri setlerinizi dönüştürmek ve geliştirmek için zenginleştirme boru hatları oluşturun ve yönetin.

Boru Hattı Detaylarını Al

GET /api/{tenantId}/{projectId}/enrichment/pipeline/{pipelineId}

Belirli bir zenginleştirme boru hattı hakkında aşamaları, yapılandırması ve yürütme meta verileri dahil olmak üzere kapsamlı bilgi alır.

Parametreler

Parametre	Tür	Konum	Açıklama
tenantId	GUID	Path	Kiracı tanımlayıcısı
projectId	GUID	Path	Proje tanımlayıcısı
pipelineId	GUID	Path	Boru hattı tanımlayıcısı

Yanıt

{
  "pipelineId": "770e8400-e29b-41d4-a716-446655440000",
  "projectId": "660e8400-e29b-41d4-a716-446655440000",
  "pipelineName": "Süreç Madenciliği Veri Zenginleştirme",
  "pipelineDescription": "Olay kayıtlarını ek özellikler ve hesaplamalarla zenginleştirir",
  "status": "Active",
  "stages": [
    {
      "stageId": "stage-001",
      "stageName": "Veri Doğrulama",
      "stageType": "Validation",
      "order": 1,
      "configuration": {
        "validateCaseId": true,
        "validateTimestamps": true,
        "requireActivityNames": true
      }
    },
    {
      "stageId": "stage-002",
      "stageName": "Zaman Zenginleştirme",
      "stageType": "TimeCalculation",
      "order": 2,
      "configuration": {
        "addDayOfWeek": true,
        "addBusinessHours": true,
        "timezoneId": "UTC"
      }
    }
  ],
  "triggers": {
    "automatic": true,
    "schedule": "0 2 * * *",
    "onDataUpdate": true
  },
  "dateCreated": "2024-01-15T10:30:00Z",
  "dateModified": "2024-01-20T14:45:00Z",
  "createdBy": "user123",
  "lastExecutionDate": "2024-01-20T02:00:00Z",
  "lastExecutionStatus": "Success",
  "executionCount": 45
}

Tüm Boru Hatlarını Listele

GET /api/{tenantId}/{projectId}/enrichment/pipelines

Projede bulunan tüm zenginleştirme boru hatlarını temel meta veri ve durum bilgisiyle birlikte getirir.

Sorgu Parametreleri

Parametre	Tür	Açıklama
status	string	Boru hattı durumuna göre filtreleme: Active, Inactive, Failed
page	integer	Sayfa numarası (varsayılan: 1)
pageSize	integer	Sayfa başına öğe sayısı (varsayılan: 20, maks: 100)

Yanıt

{
  "pipelines": [
    {
      "pipelineId": "770e8400-e29b-41d4-a716-446655440000",
      "pipelineName": "Süreç Madenciliği Veri Zenginleştirme",
      "status": "Active",
      "stageCount": 5,
      "lastExecutionDate": "2024-01-20T02:00:00Z",
      "lastExecutionStatus": "Success",
      "dateCreated": "2024-01-15T10:30:00Z"
    }
  ],
  "totalCount": 12,
  "page": 1,
  "pageSize": 20,
  "hasNextPage": false
}

Yeni Boru Hattı Oluştur

POST /api/{tenantId}/{projectId}/enrichment/pipeline

Belirtilen aşamalar ve yapılandırma ile yeni bir zenginleştirme boru hattı oluşturur. Boru hattı otomatik veya manuel çalışacak şekilde yapılandırılabilir.

İstek Gövdesi

{
  "pipelineName": "Müşteri Yolculuğu Zenginleştirme",
  "pipelineDescription": "Müşteri yolculuğu verisini demografik ve davranış kalıplarıyla zenginleştirir",
  "stages": [
    {
      "stageName": "Müşteri Veri Sorgulama",
      "stageType": "DataLookup",
      "order": 1,
      "configuration": {
        "lookupTable": "customer_demographics",
        "joinKey": "customerId",
        "selectFields": ["age", "segment", "region"]
      }
    },
    {
      "stageName": "Yolculuk Metrikleri",
      "stageType": "Calculation",
      "order": 2,
      "configuration": {
        "calculations": [
          {
            "fieldName": "journeyDuration",
            "formula": "LAST_TIMESTAMP - FIRST_TIMESTAMP",
            "groupBy": "caseId"
          },
          {
            "fieldName": "touchpointCount",
            "formula": "COUNT(*)",
            "groupBy": "caseId"
          }
        ]
      }
    }
  ],
  "triggers": {
    "automatic": false,
    "schedule": null,
    "onDataUpdate": true
  }
}

Yanıt

Oluşturulan IDs ve zaman damgaları dahil tam boru hattı nesnesi ile 201 Created döner.

Boru Hattını Güncelle

PUT /api/{tenantId}/{projectId}/enrichment/pipeline/{pipelineId}

Mevcut bir boru hattının yapılandırmasını, aşamalarını veya tetikleyicilerini günceller. Değişiklikler bir sonraki yürütmede etkili olur.

İstek Gövdesi

{
  "pipelineName": "Güncellenmiş Müşteri Yolculuğu Zenginleştirme",
  "pipelineDescription": "ML içgörüleri ile geliştirilmiş müşteri yolculuğu veri zenginleştirme",
  "status": "Active",
  "triggers": {
    "automatic": true,
    "schedule": "0 3 * * *",
    "onDataUpdate": true
  }
}

Yanıt

GET uç noktasındaki ile aynı yapıda güncellenmiş boru hattını döner.

Boru Hattını Sil

DELETE /api/{tenantId}/{projectId}/enrichment/pipeline/{pipelineId}

Bir boru hattını ve tüm yürütme geçmişini kalıcı olarak siler. Bu işlem geri alınamaz ve o anda çalışan yürütmeleri durdurur.

Yanıt Kodları

• 204 No Content - Boru hattı başarıyla silindi
• 404 Not Found - Boru hattı bulunamadı veya erişim reddedildi
• 409 Conflict - Boru hattı şu anda çalışıyor ve silinemez

Boru Hattına Aşama Ekle

POST /api/{tenantId}/{projectId}/enrichment/pipeline/{pipelineId}/stage

Mevcut bir boru hattına yeni bir işlem aşaması ekler. Aşama belirtilen sıra konumuna yerleştirilecektir.

İstek Gövdesi

{
  "stageName": "Süreç Performans Metrikleri",
  "stageType": "PerformanceCalculation",
  "order": 3,
  "configuration": {
    "metrics": [
      {
        "name": "cycleTime",
        "calculation": "CASE_DURATION",
        "unit": "hours"
      },
      {
        "name": "waitTime",
        "calculation": "ACTIVITY_WAITING_TIME",
        "unit": "hours"
      }
    ],
    "aggregations": ["AVG", "MAX", "MIN", "P95"]
  }
}

Yanıt

Oluşturulan aşama ID'si dahil tamamlanmış aşama nesnesi ile 201 Created döner.

Boru Hattından Aşama Kaldır

DELETE /api/{tenantId}/{projectId}/enrichment/pipeline/{pipelineId}/stage/{stageId}

Boru hattından belirli bir aşamayı kaldırır. Takip eden aşamalar otomatik olarak yeniden sıralanır.

Yanıt Kodları

• 204 No Content - Aşama başarıyla kaldırıldı
• 404 Not Found - Aşama boru hattında bulunamadı
• 409 Conflict - Boru hattı çalışırken aşama kaldırılamaz

Örnek: Tam Boru Hattı İş Akışı

Bu örnek, bir zenginleştirme boru hattı oluşturmayı ve yönetmeyi gösterir:

// 1. Yeni bir zenginleştirme boru hattı oluştur
const createPipeline = async () => {
  const response = await fetch('/api/{tenantId}/{projectId}/enrichment/pipeline', {
    method: 'POST',
    headers: {
      'Content-Type': 'application/json',
      'Authorization': `Bearer ${token}`
    },
    body: JSON.stringify({
      pipelineName: 'Sipariş İşleme Zenginleştirme',
      pipelineDescription: 'Sipariş verisini gerçekleştirme metrikleri ile zenginleştirir',
      stages: [
        {
          stageName: 'Sipariş Doğrulama',
          stageType: 'Validation',
          order: 1,
          configuration: {
            validateOrderId: true,
            validateCustomerId: true,
            validateAmounts: true
          }
        },
        {
          stageName: 'Gerçekleşme Süresi Hesaplama',
          stageType: 'TimeCalculation',
          order: 2,
          configuration: {
            startActivity: 'Order Received',
            endActivity: 'Order Shipped',
            outputField: 'fulfillmentTime'
          }
        }
      ],
      triggers: {
        automatic: true,
        onDataUpdate: true
      }
    })
  });

  return await response.json();
};

// 2. Mevcut boru hattına yeni bir aşama ekle
const addStage = async (pipelineId) => {
  const response = await fetch(`/api/{tenantId}/{projectId}/enrichment/pipeline/${pipelineId}/stage`, {
    method: 'POST',
    headers: {
      'Content-Type': 'application/json',
      'Authorization': `Bearer ${token}`
    },
    body: JSON.stringify({
      stageName: 'Müşteri Segmentasyonu',
      stageType: 'Classification',
      order: 3,
      configuration: {
        segmentationRules: [
          {
            segment: 'VIP',
            condition: 'orderValue > 1000'
          },
          {
            segment: 'Regular',
            condition: 'orderValue <= 1000'
          }
        ]
      }
    })
  });

  return await response.json();
};

// 3. Boru hattı durumunu al
const getPipelineStatus = async (pipelineId) => {
  const response = await fetch(`/api/{tenantId}/{projectId}/enrichment/pipeline/${pipelineId}`, {
    headers: {
      'Authorization': `Bearer ${token}`
    }
  });

  return await response.json();
};

Python Örneği

import requests
import json
from datetime import datetime

class EnrichmentPipelineManager:
    def __init__(self, base_url, tenant_id, project_id, token):
        self.base_url = base_url
        self.tenant_id = tenant_id
        self.project_id = project_id
        self.headers = {
            'Authorization': f'Bearer {token}',
            'Content-Type': 'application/json'
        }

    def create_pipeline(self, name, description, stages, triggers=None):
        """Yeni bir zenginleştirme boru hattı oluştur"""
        url = f"{self.base_url}/api/{self.tenant_id}/{self.project_id}/enrichment/pipeline"
        payload = {
            'pipelineName': name,
            'pipelineDescription': description,
            'stages': stages,
            'triggers': triggers or {'automatic': False, 'onDataUpdate': True}
        }
        response = requests.post(url, json=payload, headers=self.headers)
        return response.json()

    def get_pipeline(self, pipeline_id):
        """Boru hattı detaylarını al"""
        url = f"{self.base_url}/api/{self.tenant_id}/{self.project_id}/enrichment/pipeline/{pipeline_id}"
        response = requests.get(url, headers=self.headers)
        return response.json()

    def list_pipelines(self, status=None, page=1, page_size=20):
        """Tüm boru hatlarını isteğe bağlı filtreleme ile listele"""
        url = f"{self.base_url}/api/{self.tenant_id}/{self.project_id}/enrichment/pipelines"
        params = {'page': page, 'pageSize': page_size}
        if status:
            params['status'] = status

        response = requests.get(url, params=params, headers=self.headers)
        return response.json()

    def add_stage(self, pipeline_id, stage_name, stage_type, order, configuration):
        """Mevcut boru hattına yeni aşama ekle"""
        url = f"{self.base_url}/api/{self.tenant_id}/{self.project_id}/enrichment/pipeline/{pipeline_id}/stage"
        payload = {
            'stageName': stage_name,
            'stageType': stage_type,
            'order': order,
            'configuration': configuration
        }
        response = requests.post(url, json=payload, headers=self.headers)
        return response.json()

    def update_pipeline(self, pipeline_id, name=None, description=None, status=None, triggers=None):
        """Boru hattı yapılandırmasını güncelle"""
        url = f"{self.base_url}/api/{self.tenant_id}/{self.project_id}/enrichment/pipeline/{pipeline_id}"
        payload = {}
        if name:
            payload['pipelineName'] = name
        if description:
            payload['pipelineDescription'] = description
        if status:
            payload['status'] = status
        if triggers:
            payload['triggers'] = triggers

        response = requests.put(url, json=payload, headers=self.headers)
        return response.json()

    def delete_pipeline(self, pipeline_id):
        """Bir boru hattını sil"""
        url = f"{self.base_url}/api/{self.tenant_id}/{self.project_id}/enrichment/pipeline/{pipeline_id}"
        response = requests.delete(url, headers=self.headers)
        return response.status_code == 204

# Kullanım örneği
manager = EnrichmentPipelineManager(
    'https://your-mindzie-instance.com',
    'tenant-guid',
    'project-guid',
    'your-auth-token'
)

# Kapsamlı bir zenginleştirme boru hattı oluştur
stages = [
    {
        'stageName': 'Veri Kalitesi Kontrolü',
        'stageType': 'Validation',
        'order': 1,
        'configuration': {
            'checkDuplicates': True,
            'validateTimestamps': True,
            'checkMissingValues': True
        }
    },
    {
        'stageName': 'Süreç Madenciliği Metrikleri',
        'stageType': 'ProcessCalculation',
        'order': 2,
        'configuration': {
            'calculateCycleTime': True,
            'calculateWaitingTime': True,
            'calculateResourceUtilization': True,
            'detectBottlenecks': True
        }
    },
    {
        'stageName': 'Anomali Tespiti',
        'stageType': 'AnomalyDetection',
        'order': 3,
        'configuration': {
            'algorithm': 'isolation_forest',
            'threshold': 0.1,
            'features': ['duration', 'cost', 'resourceCount']
        }
    }
]

pipeline = manager.create_pipeline(
    'Kapsamlı Süreç Analizi',
    'Uçtan uca süreç analizi ve anomali tespiti',
    stages,
    {'automatic': True, 'schedule': '0 1 * * *', 'onDataUpdate': True}
)

print(f"Oluşturulan boru hattı: {pipeline['pipelineId']}")

# Tüm aktif boru hatlarını listele
active_pipelines = manager.list_pipelines(status='Active')
print(f"{active_pipelines['totalCount']} adet aktif boru hattı bulundu")