$changeStreamSplitLargeEvent (estágio de agregação )

Definição

$changeStreamSplitLargeEvent

Novidade no MongoDB 7.0 (e 6.0.9).

Se umchange stream tiver eventos grandes que excedam 16 MB, uma exceção BSONObjectTooLarge será retornada. A partir do MongoDB 7.0 (e 6.0.9), você pode utilizar um estágio $changeStreamSplitLargeEvent para divisão os eventos em fragmentos menores.

Você só deve usar $changeStreamSplitLargeEvent quando estritamente necessário. Por exemplo, se seu aplicativo exigir imagens pré ou pós-documento completo e gerar eventos grandes que excedam 16 MB, use $changeStreamSplitLargeEvent.

Antes de decidir usar $changeStreamSplitLargeEvent, você deve primeiro tentar reduzir o tamanho do evento de alteração. Por exemplo:

• Não solicite pré ou pós-imagens de documentos, a menos que seu aplicativo as exija. Isso gera campos fullDocument e fullDocumentBeforeChange em mais casos, que normalmente são os maiores objetos em um evento de alteração.

• Use um estágio $project a fim de incluir apenas os campos necessários para sua aplicação. Isso reduz o tamanho do evento de alteração e evita o tempo adicional para dividir eventos grandes em fragmentos. Assim, mais eventos de alteração podem ser retornados em cada lote.

Você só pode ter um estágio $changeStreamSplitLargeEvent no seu pipeline, e ele deve ser o último. Você só pode usar $changeStreamSplitLargeEvent em um pipeline $changeStream.

$changeStreamSplitLargeEvent sintaxe:

{
  $changeStreamSplitLargeEvent: {}
}

Comportamento

$changeStreamSplitLargeEvent divide eventos que excedem 16 MB em fragmentos e retorna os fragmentos sequencialmente usando o cursor de fluxo de alterações.

Os fragmentos são divisão para que o número máximo de campos seja retornado no primeiro fragmento. Isso garante que o contexto do evento seja retornado o mais rápido possível.

Quando o evento de alteração é dividido, somente o tamanho dos campos de nível superior é usado. $changeStreamSplitLargeEvent não processa nem faz a divisão de subdocumentos recursivamente. Por exemplo, se você usar um estágio $project para criar um evento de alteração com um único campo que tem 20 MB de tamanho, o evento não sofrerá uma divisão, e o estágio retornará um erro.

Cada fragmento possui um token de retomada. Um stream retomado usando o token de um fragmento irá:

• Iniciar um novo fluxo a partir do fragmento subsequente.

• Comece no próximo evento se continuar a partir do fragmento final da sequência.

Cada fragmento de um evento inclui um documento splitEvent :

splitEvent: {
   fragment: <int>,
   of: <int>
}

A tabela a seguir descreve os campos.

Exemplos

db.myCollection.insertOne(
   { _id: 0, largeField: "a".repeat( 16 * 1024 * 1024 - 1024 ) }
)

db.runCommand( {
   collMod: "myCollection",
   changeStreamPreAndPostImages: { enabled: true }
} )

myChangeStreamCursor = db.myCollection.watch(
   [ { $changeStreamSplitLargeEvent: {} } ],
   { fullDocument: "required", fullDocumentBeforeChange: "required" }
)

db.myCollection.updateOne(
   { _id: 0 },
   { $set: { largeField: "b".repeat( 16 * 1024 * 1024 - 1024 ) } }
)

const firstFragment = myChangeStreamCursor.next()
const secondFragment = myChangeStreamCursor.next()
const thirdFragment = myChangeStreamCursor.next()

firstFragment.splitEvent

splitEvent: { fragment: 1, of: 3 }

splitEvent: { fragment: 2, of: 3 }
splitEvent: { fragment: 3, of: 3 }

Object.keys( firstFragment )

[
   '_id',
   'splitEvent',
   'wallTime',
   'clusterTime',
   'operationType',
   'documentKey',
   'ns',
   'fullDocument'
]

bsonsize( firstFragment.fullDocument )

16776223

Object.keys( secondFragment )

[ '_id', 'splitEvent', 'fullDocumentBeforeChange' ]

Object.keys( thirdFragment )

[ '_id', 'splitEvent', 'updateDescription' ]

[BsonIgnoreExtraElements]
public class LargeDocument
{
    [BsonId]
    public int Id { get; set; }
    public string LargeField { get; set; } = null!;
}

var client = new MongoClient("<connection-string>");
var db = client.GetDatabase("change_stream_test");
var collection = db.GetCollection<LargeDocument>("largeEventCollection");
collection.InsertOne(new LargeDocument
{
    Id = 0,
    LargeField = new string('a', 16 * 1024 * 1024 - 1024)
});
db.RunCommand<BsonDocument>(new BsonDocument
{
    { "collMod", "largeEventCollection" },
    { "changeStreamPreAndPostImages", new BsonDocument("enabled", true) }
});
var pipeline = new EmptyPipelineDefinition<ChangeStreamDocument<LargeDocument>>()
    .ChangeStreamSplitLargeEvent();
var watchOptions = new ChangeStreamOptions
{
    FullDocument = ChangeStreamFullDocumentOption.Required,
    FullDocumentBeforeChange = ChangeStreamFullDocumentBeforeChangeOption.Required
};
using var cursor = collection.Watch(pipeline, watchOptions);
collection.UpdateOne(
    Builders<LargeDocument>.Filter.Eq(d => d.Id, 0),
    Builders<LargeDocument>.Update.Set(d => d.LargeField, new string('b', 16 * 1024 * 1024 - 1024))
);
while (cursor.MoveNext())
{
    foreach (var fragment in cursor.Current)
    {
        Console.WriteLine(fragment.BackingDocument["splitEvent"].ToJson());
    }
}

{ "fragment" : 1, "of" : 3 }
{ "fragment" : 2, "of" : 3 }
{ "fragment" : 3, "of" : 3 }

const pipeline = [{ changeStreamSplitLargeEvent: {} }];
const changeStream = collection.watch(pipeline);
return changeStream;

Saiba mais

Para mais informações sobre como alterar as notificações de transmissão, consulte Alterar eventos.

Para saber mais sobre os estágios de pipeline relacionados, consulte o guia$changeStream.