Logo
Şema eğitimi
Şema eğitimiDers 28: Büyük veri kümelerini güncelleme

Ders 28: Büyük veri kümelerini güncelleme

Bazen tek bir işlemde binlerce kaynağı güncellememiz gerekir; aşağıdaki yorum bunu güzel bir şekilde ifade etmektedir (WordPress hakkında bir topluluk grubunda yayımlanmıştır):

Birlikte çalıştığım pek çok müşteri için büyük veri kümeleriyle (1 ürün için 10.000'den fazla ürün varyasyonu veya 13.000'den fazla medya dosyası) çalışmam gerektiğini görüyorum... Müşteriler kaçınılmaz olarak birçok şeyi aynı anda toplu düzenleyebilmek istiyor — örneğin 2.000 medya dosyasını aynı etiketle etiketlemek gibi.

Bu tutorial dersinde bu görevi nasıl ele alabileceğimizi inceleyeceğiz.

Nested Mutations

Bu GraphQL query'sinin çalışması için, endpoint'e uygulanan Şema Yapılandırması'nın Nested Mutations özelliğinin etkinleştirilmiş olması gerekir

Nested Mutations sayesinde, tek bir GraphQL query'si aracılığıyla veritabanından binlerce kaynağı alabilir ve güncelleyebiliriz:

mutation ReplaceOldWithNewDomainInPosts {
  posts(pagination: { limit: 3000 }) {
    id
    rawContent
    adaptedRawContent: _strReplace(
      search: "https://my-old-domain.com"
      replaceWith: "https://my-new-domain.com"
      in: $__rawContent
    )
    update(input: {
      contentAs: { html: $__adaptedRawContent }
    }) {
      status
      errors {
        __typename
        ...on ErrorPayload {
          message
        }
      }
    }
  }
}

Ancak sistemin dayanıklılığına bağlı olarak, bu tek GraphQL çalıştırması veritabanına aşırı yük bindirerek çökmesine bile neden olabilir.

GraphQL query'sinin çalıştırılmasını sayfalandırma

Binlerce kaynağı aynı anda güncellemek sistemi çökertiyorsa çözüm basittir: GraphQL'i binlerce kaynak için tek seferde çalıştırmak yerine, her seferinde onlarca kaynak işleyerek yüzlerce kez çalıştırabiliriz.

Aşağıdaki bash betikleri önce commentCount aracılığıyla toplam yorum sayısını bulur, ardından $ENTRIES_TO_PROCESS ortam değişkenini dikkate alarak segmentleri hesaplar ve her segment için sayfalandırma parametrelerini hesaplayıp GraphQL query'sini çağırır (yalnızca o segmentteki yorumları alır):

# Get the number of comments in the site
GRAPHQL_RESPONSE=$(curl
  -X POST \
  -H "Content-Type: application/json" \
  -d '{"query": "{\n  commentCount\n}"}' \
  https://mysite.com/graphql/)
 
# Extract the number of comments into a variable
COMMENT_COUNT=$(echo $GRAPHQL_RESPONSE \
  | grep -E -o '"commentCount\":([0-9]+)' \
  | cut -d':' -f2-)
 
echo "Number of comments: $COMMENT_COUNT"
 
# How many entries will be processed on each query
ENTRIES_TO_PROCESS=10
 
# Calculate how many requests must be triggered
PAGINATION_COUNT=$(($(($COMMENT_COUNT / $ENTRIES_TO_PROCESS)) + $(($(($COMMENT_COUNT % $ENTRIES_TO_PROCESS)) ? 1 : 0))))
 
echo "Number of requests to process (at $ENTRIES_TO_PROCESS entries per request): $PAGINATION_COUNT"
 
# Execute the requests, at one per second
for PAGINATION_NUMBER in $(seq 0 $(($PAGINATION_COUNT - 1))); do sleep 1 && echo "\n\nPagination number: $PAGINATION_NUMBER\n" && curl -X POST -H "Content-Type: application/json" -d "{\"query\": \"{ comments(pagination: { limit: $ENTRIES_TO_PROCESS, offset: $(($PAGINATION_NUMBER * $ENTRIES_TO_PROCESS)) }) { id date content } }\"}" https://mysite.com/graphql/ ; done

GraphQL query'sini özyinelemeli olarak çalıştırma

Yukarıdaki çözüm bash betikleri gerektirdiğinden, yalnızca CLI (veya bir yönetici paneli ya da araç) aracılığıyla çalıştırılabilir; bu da kullanımını kısıtlar.

Aynı mantığı doğrudan GraphQL query'sinin kendisine yansıtabilir, böylece bunu zaten WordPress içinde çalıştırabiliriz (hatta bir GraphQL Persisted Query olarak kaydedebiliriz).

Aşağıdaki GraphQL query'si kendisini özyinelemeli olarak çalıştırır. İlk çağrıldığında:

  • Güncellenecek toplam kaynak sayısını segmentlere böler (sağlanan $limit değişkeni kullanılarak hesaplanır)
  • Her segment için yeni bir HTTP isteği aracılığıyla kendisini çalıştırır (karşılık gelen $offset'i değişken olarak ileterek), böylece belirli bir anda yalnızca tüm kaynakların bir alt kümesini günceller

GraphQL query'si, HTTP isteklerinin geçerli URL'nin aynısına işaret etmesi (artı o segment için $offset değişkeninin eklenmesi) sayesinde özyinelemeli hale gelir; bu URL'yi (ve ayrıca gövde, yöntem ve başlıkları) geçerli HTTP isteğinden alırız (HTTP Request via Schema uzantısı aracılığıyla).

_sendHTTPRequests'e iletilen $async argümanı false olarak ayarlanmıştır; bu sayede HTTP istekleri birbiri ardına çalıştırılır. Ayrıca isteğe bağlı $delay değişkeni, her istek gönderilmeden önce kaç milisaniye bekleneceğini belirtmeye olanak tanır.

Tüm kaynaklar güncellendikten sonra GraphQL query'sinin çalıştırılması sona erer ve işlem tamamlanır:

# When first invoked, we do not pass variable `$offset`
# Then `$offset` is `null`, and dynamic variable `$executeQuery` will be `true`
query ExportExecute(
  $offset: Int
) {
  executeQuery: _notNull(value: $offset)
    @export(as: "executeQuery")
    @remove # Comment this directive to visualize output during development
}
 
# Only calculate the segments on the first invocation of the GraphQL query
query CalculateVars($limit: Int! = 10)
  @depends(on: "ExportExecute")
  @skip(if: $executeQuery)
{
  # Calculate the number of HTTP requests to be sent
  commentCount
  fractionalNumberExecutions: _floatDivide(number: $__commentCount, by: $limit)
    @remove # Comment this directive to visualize output during development
  numberExecutions: _floatCeil(number: $__fractionalNumberExecutions)
  
  # Generate a list of the offset
  arrayOffsets: _arrayPad(array: [], length: $__numberExecutions, value: null)
    @underEachArrayItem(
      passIndexOnwardsAs: "position"
    )
      @applyField(
        name: "_intMultiply"
        arguments: {
          multiply: $position
          with: $limit
        }
        setResultInResponse: true
      )
    @export(as: "offsets")
 
  # Vars needed to generate a list of the HTTP Request inputs,
  # with many of them retrieved from the current HTTP request data
  url: _httpRequestFullURL
    @export(as: "url")
    @remove # Comment this directive to visualize output during development
  method: _httpRequestMethod
    @export(as: "method")
    @remove # Comment this directive to visualize output during development
  headers: _httpRequestHeaders
    @remove # Comment this directive to visualize output during development
  headersInputList: _objectConvertToNameValueEntryList(
    object: $__headers
  )
    @export(as: "headersInputList")
    @remove # Comment this directive to visualize output during development
  body: _httpRequestBody
    @remove # Comment this directive to visualize output during development
  bodyJSONObject: _strDecodeJSONObject(string: $__body)
    @export(as: "bodyJSONObject")
    @remove # Comment this directive to visualize output during development
  bodyHasVariables: _propertyIsSetInJSONObject(
    object: $__bodyJSONObject,
    by: { key: "variables" }
  )
    @export(as: "bodyHasVariables")
    @remove # Comment this directive to visualize output during development
}
 
query GenerateVars
  @depends(on: ["ExportExecute", "CalculateVars"])
  @skip(if: $executeQuery)
{
  bodyJSON: _echo(value: $bodyJSONObject)
    @unless(condition: $bodyHasVariables)
      @objectAddEntry(
        key: "variables"
        value: {}
      )
    @export(as: "bodyJSON")
    @remove # Comment this directive to visualize output during development
}
 
# Generate all the HTTPRequestInput objects to send each of the HTTP requests
query GenerateRequestInputs(
  $timeout: Float,
  $delay: Int
)
  @depends(on: ["ExportExecute", "GenerateVars"])
  @skip(if: $executeQuery)
{
  # Generate a list of the HTTP Request inputs (without the offset)
  requestInputs: _echo(value: $offsets)
    @underEachArrayItem(
      passValueOnwardsAs: "requestOffset"
      affectDirectivesUnderPos: [1, 2]
    )
      @applyField(
        name: "_objectAddEntry",
        arguments: {
          object: $bodyJSON
          underPath: "variables"
          key: "offset"
          value: $requestOffset
        },
        passOnwardsAs: "itemJSON"
      )
      @applyField(
        name: "_echo",
        arguments: {
          value: {
            url: $url
            method: $method
            options: {
              headers: $headersInputList
              json: $itemJSON
              timeout: $timeout
              delay: $delay
            }
          }
        },
        setResultInResponse: true
      )
    @export(as: "requestInputs")
    @remove # Comment this directive to visualize output during development
}
 
# Execute all the generated URLs, either asynchronously or not
query ExecuteURLs
  @depends(on: ["ExportExecute", "GenerateRequestInputs"])
  @skip(if: $executeQuery)
{
  _sendHTTPRequests(
    async: false
    inputs: $requestInputs
  ) {
    statusCode
    contentType
    body
      @remove
    bodyJSON: _strDecodeJSONObject(string: $__body)
  }
}
 
# This is the actual execution of the query.
# In this case, it simply prints the time when it was executed,
# the provided query variables, and the comment IDs for that segment
query ExecuteQuery(
  $offset: Int
  $limit: Int! = 10
)
  @depends(on: "ExportExecute")
  @include(if: $executeQuery)
{
  executionTime: _httpRequestRequestTime
  queryVariables: _sprintf(string: "[$limit: %s, $offset: %s]", values: [$limit, $offset])
  comments(
    pagination: { limit: $limit, offset: $offset }
    sort: { order: ASC, by: ID }
  ) {
    id
  }
}
 
query ExecuteAll
  @depends(on: ["ExecuteURLs", "ExecuteQuery"])
{
  id
    @remove
}

Yanıt şu şekildedir:

{
  "data": {
    "commentCount": 23,
    "numberExecutions": 3,
    "arrayOffsets": [
      0,
      10,
      20
    ],
    "_sendHTTPRequests": [
      {
        "statusCode": 200,
        "contentType": "application/json",
        "bodyJSON": {
          "data": {
            "executionTime": 1689814467,
            "queryVariables": "[$limit: 10, $offset: 0]",
            "comments": [
              {
                "id": 2
              },
              {
                "id": 3
              },
              {
                "id": 4
              },
              {
                "id": 5
              },
              {
                "id": 6
              },
              {
                "id": 7
              },
              {
                "id": 8
              },
              {
                "id": 9
              },
              {
                "id": 10
              },
              {
                "id": 11
              }
            ]
          }
        }
      },
      {
        "statusCode": 200,
        "contentType": "application/json",
        "bodyJSON": {
          "data": {
            "executionTime": 1689814468,
            "queryVariables": "[$limit: 10, $offset: 10]",
            "comments": [
              {
                "id": 12
              },
              {
                "id": 13
              },
              {
                "id": 16
              },
              {
                "id": 17
              },
              {
                "id": 18
              },
              {
                "id": 19
              },
              {
                "id": 20
              },
              {
                "id": 21
              },
              {
                "id": 22
              },
              {
                "id": 23
              }
            ]
          }
        }
      },
      {
        "statusCode": 200,
        "contentType": "application/json",
        "bodyJSON": {
          "data": {
            "executionTime": 1689814470,
            "queryVariables": "[$limit: 10, $offset: 20]",
            "comments": [
              {
                "id": 24
              },
              {
                "id": 25
              },
              {
                "id": 26
              }
            ]
          }
        }
      }
    ]
  }
}