Umgang mit gedrosselten Anrufen und gelöschten

Ein Amazon Textract Textract-Vorgang kann fehlschlagen, wenn Sie die maximale Anzahl von Transaktionen pro Sekunde (TPS) überschreiten, wodurch der Service Ihre Anwendung drosselt oder wenn Ihre Verbindung unterbrochen wird. Wenn Sie beispielsweise in kurzer Zeit zu viele Anrufe Amazon Textract Textract-Operationen tätigen, drosselt es Ihre Anrufe und sendet eineProvisionedThroughputExceededExceptionFehler in der Betriebsantwort. Hinweise zu Amazon Textract TPS-Kontingenten finden Sie unterAmazon Textract Kontingenteaus.

Sie können Drosselung und abgebrochene Verbindungen verwalten, indem Sie den Vorgang automatisch wiederholen. Sie können die Anzahl der Wiederholungen angeben, indem Sie dieConfig-Parameter, wenn Sie den Amazon Textract Textract-Client erstellen. Wir empfehlen eine Wiederholungsanzahl von 5. DieAWSDas SDK versucht eine Operation die angegebene Anzahl von Malen erneut, bevor es fehlschlägt und eine Ausnahme ausgelöst wird. Weitere Informationen finden Sie unter Wiederholversuche bei Fehlern und exponentielles Backoff in AWS.

Anmerkung

Automatische Wiederholungen funktionieren sowohl für synchrone als auch für asynchrone Operationen. Stellen Sie vor dem Angeben von automatischen Wiederholungen sicher, dass Sie über die neueste Version des AWS SDKs verfügen. Weitere Informationen finden Sie unter Schritt 2: Einrichten derAWS CLIundAWS-SDKs .

Das folgende Beispiel zeigt, wie Sie Amazon Textract Textract-Vorgänge automatisch erneut versuchen, wenn Sie mehrere Dokumente verarbeiten.

Voraussetzungen

• Wenn Sie dies noch nicht getan haben:

  1. Erstellen oder aktualisieren Sie einen IAM-Benutzer mitAmazonTextractFullAccessundAmazonS3ReadOnlyAccess-Berechtigungen Weitere Informationen finden Sie unter Schritt 1: Einrichten eines AWS-Kontos und Erstellen eines IAM-Benutzers .

  2. Installieren und konfigurieren Sie die AWS CLI- und AWS-SDKs. Weitere Informationen finden Sie unter Schritt 2: Einrichten derAWS CLIundAWS-SDKs .

So wiederholen Sie Operationen automatisch

1. Laden Sie mehrere Dokumentbilder in Ihren S3-Bucket hoch, um das Synchrone Beispiel auszuführen. Laden Sie ein mehrseitiges Dokument in Ihren S3-Bucket hoch und führen Sie esStartDocumentTextDetectiondarauf, um das Asynchrone Beispiel auszuführen.

   Detaillierte Anweisungen finden Sie unterHochladen von Objekten in Amazon S3imAmazon Simple Storage Service Benutzerhandbuchaus.

2. Die folgenden Beispiele veranschaulichen, wie Sie dieConfig-Parameter, um eine Operation automatisch erneut zu versuchen. Das synchrone Beispiel ruft denDetectDocumentText-Operation, während das Asynchrone Beispiel dieGetDocumentTextDetectionverwenden.

   Sync Example

   Verwenden Sie die folgenden Beispiele, um dieDetectDocumentText-Betrieb auf den Dokumenten in Ihrem Amazon S3 S3-Bucket. In :mainSo ändern Sie den -Wertbucketauf Ihrem S3-Bucket hoch (es sollte ein oder mehrere Objekte wie beispielsweise Bäume, Häuser, Boote, etc. enthalten). Ändern Sie den -Wertdocumentsauf die Namen der Dokumentbilder, die Sie in Schritt 2 hochgeladen haben.

   import boto3
   from botocore.client import Config
   # Documents
   
   def process_multiple_documents(bucket, documents):
       
       config = Config(retries = dict(max_attempts = 5))
    
       # Amazon Textract client
       textract = boto3.client('textract', config=config)
    
       for documentName in documents:
    
           print("\nProcessing: {}\n==========================================".format(documentName))
    
           # Call Amazon Textract
           response = textract.detect_document_text(
               Document={
                   'S3Object': {
                       'Bucket': bucket,
                       'Name': documentName
                   }
               })
    
           # Print detected text
           for item in response["Blocks"]:
               if item["BlockType"] == "LINE":
                   print ('\033[94m' +  item["Text"] + '\033[0m')
   
   
   def main():
       bucket = ""
       documents = ["document-image-1.png",
       "document-image-2.png", "document-image-3.png",
       "document-image-4.png", "document-image-5.png" ]
       process_multiple_documents(bucket, documents)
   
   
   
   if __name__ == "__main__":
       main()

   Async Example

   Verwenden Sie die folgenden Beispiele zum Aufrufen der GetDocumentTextDetection-Operation. Es wird davon ausgegangen, dass Sie bereits angerufen habenStartDocumentTextDetectionauf den Dokumenten in Ihrem Amazon S3 S3-Bucket erhalten und eineJobIdaus. In :mainSo ändern Sie den -Wertbucketauf Ihrem S3-Bucket und dem -WertroleArnzu dem Arn, der Ihrer Textrakt-Rolle zugewiesen wurde. Sie müssen auch den -Wert änderndocumentauf den Namen Ihres mehrseitigen Dokuments in Ihrem Amazon S3 S3-Bucket hoch. Ersetzen Sie schließlich den -Wertregion_nameGeben Sie den Namen Ihrer Region an und geben Sie dieGetResults-Funktion mit dem Namen IhresjobIdaus.

   import boto3
   from botocore.client import Config
   
   class DocumentProcessor:
       jobId = ''
       region_name = ''
   
       roleArn = ''
       bucket = ''
       document = ''
   
       sqsQueueUrl = ''
       snsTopicArn = ''
       processType = ''
   
       def __init__(self, role, bucket, document, region):
           self.roleArn = role
           self.bucket = bucket
           self.document = document
           self.region_name = region
           self.config = Config(retries = dict(max_attempts = 5))
   
           self.textract = boto3.client('textract', region_name=self.region_name, config=self.config)
           self.sqs = boto3.client('sqs')
           self.sns = boto3.client('sns')
   
   # Display information about a block
       def DisplayBlockInfo(self, block):
   
           print("Block Id: " + block['Id'])
           print("Type: " + block['BlockType'])
           if 'EntityTypes' in block:
               print('EntityTypes: {}'.format(block['EntityTypes']))
   
           if 'Text' in block:
               print("Text: " + block['Text'])
   
           if block['BlockType'] != 'PAGE':
               print("Confidence: " + "{:.2f}".format(block['Confidence']) + "%")
   
           print('Page: {}'.format(block['Page']))
   
           if block['BlockType'] == 'CELL':
               print('Cell Information')
               print('\tColumn: {} '.format(block['ColumnIndex']))
               print('\tRow: {}'.format(block['RowIndex']))
               print('\tColumn span: {} '.format(block['ColumnSpan']))
               print('\tRow span: {}'.format(block['RowSpan']))
   
               if 'Relationships' in block:
                   print('\tRelationships: {}'.format(block['Relationships']))
   
           print('Geometry')
           print('\tBounding Box: {}'.format(block['Geometry']['BoundingBox']))
           print('\tPolygon: {}'.format(block['Geometry']['Polygon']))
   
           if block['BlockType'] == 'SELECTION_ELEMENT':
               print('    Selection element detected: ', end='')
               if block['SelectionStatus'] == 'SELECTED':
                   print('Selected')
               else:
                   print('Not selected')
   
       def GetResults(self, jobId):
           maxResults = 1000
           paginationToken = None
           finished = False
   
           while finished == False:
   
               response = None
   
               if paginationToken == None:
                   response = self.textract.get_document_text_detection(JobId=jobId,
                                                                            MaxResults=maxResults)
               else:
                   response = self.textract.get_document_text_detection(JobId=jobId,
                                                                            MaxResults=maxResults,
                                                                            NextToken=paginationToken)
   
               blocks = response['Blocks']
               print('Detected Document Text')
               print('Pages: {}'.format(response['DocumentMetadata']['Pages']))
   
               # Display block information
               for block in blocks:
                   self.DisplayBlockInfo(block)
                   print()
                   print()
   
               if 'NextToken' in response:
                   paginationToken = response['NextToken']
               else:
                   finished = True
   
   def main():
       roleArn = 'role-arn'
       bucket = 'bucket-name'
       document = 'document-name'
       region_name = 'region-name'
       analyzer = DocumentProcessor(roleArn, bucket, document, region_name)
       analyzer.GetResults("job-id")
   
   if __name__ == "__main__":
       main()