Schema 教程第28课：批量更新大规模数据
第28课：批量更新大规模数据
有时我们需要在单次操作中更新数千个资源，正如以下评论（发布在一个关于 WordPress 的社区群组中）所描述的那样：
我发现对于很多客户，我都在处理大量数据（1个产品有10,000多个变体，或者13,000多个媒体文件）……客户不可避免地希望能够批量编辑大量内容，比如为2,000个媒体文件打上同一个标签。
在本教程课程中，我们将探索解决这一任务的方法。
Nested Mutations
要使此 GraphQL query 正常工作，应用于端点的架构配置需要启用 Nested Mutations
得益于 Nested Mutations，我们可以通过单个 GraphQL query 从数据库中检索并更新数千个资源：
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
}
}
}
}
}然而，根据系统的承载能力，这单次 GraphQL 执行可能会给数据库带来过大负荷，甚至导致其崩溃。
对 GraphQL query 的执行进行分页
如果一次性更新数千个资源会导致系统崩溃，解决方案很简单：不必为数千个资源只执行一次 GraphQL，我们可以将其分成数百次执行，每次处理数十个资源。
以下 bash 脚本首先通过 commentCount 获取评论总数，然后根据环境变量 $ENTRIES_TO_PROCESS 计算分段，并为每个分段计算分页参数并调用 GraphQL query（仅检索该分段的评论）：
# 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
由于上述解决方案涉及 bash 脚本，必须通过 CLI（或某个管理面板或工具）执行，限制了其使用范围。
我们可以将相同的逻辑嵌入到 GraphQL query 本身中，从而可以直接在 WordPress 内执行（甚至可以将其存储为 GraphQL Persisted Query）。
下面的 GraphQL query 会递归地执行自身。首次调用时，它会：
- 将需要更新的资源总数划分为若干分段（使用提供的
$limit变量计算） - 为每个分段通过新的 HTTP 请求执行自身（将对应的
$offset作为变量传递），从而在任意时刻只更新部分资源
该 GraphQL query 通过将 HTTP 请求指向与当前相同的 URL（加上该分段的 $offset 变量）来实现递归，我们从当前 HTTP 请求中获取该 URL（以及请求体、方法和头部信息），使用的是 HTTP Request via Schema 扩展。
传递给 _sendHTTPRequests 的 $async 参数已设置为 false，因此 HTTP 请求将依次执行。此外，可选变量 $delay 允许指定在发送每个请求前延迟的毫秒数。
所有资源更新完毕后，GraphQL query 的执行到达末尾并终止：
# 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
}响应如下：
{
"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
}
]
}
}
}
]
}
}