GetRecords

On This Page

Description

Retrieves (consumes) records from a stream shard.

Determining the Start Location

Before submitting a GetRecords request, you need to determine the location within the shard at which to begin the record consumption. Before the first request for a specific shard, send a Seek request to get the desired location (see the Seek Location) response element. In subsequent GetRecords requests, pass the NextLocation value that you received in a previous GetRecords response as the Location value for the current request. See the descriptions of Location and NextLocation, and the Stream Record Consumption overview.

Request

Request Header

Syntax
    POST /<container>/<resource> HTTP/1.1
Host: <web-APIs URL>
Content-Type: application/json
X-v3io-function: GetRecords
<Authorization OR X-v3io-session-key>: <value>

    url = "http://<web-APIs URL>/<container>/<resource>"
headers = {
            "Content-Type": "application/json",
            "X-v3io-function": "GetRecords",
            "<Authorization OR X-v3io-session-key>": "<value>"
          }
    URL Resource Parameters

    The path to the stream shard from which to retrieve the records. The path includes the stream path and the shard ID. You can optionally set the stream name and shard ID, or only the shard ID, in the request's StreamName and ShardId JSON parameters instead of in the URL.

    Request Data

    Syntax
      {
    "StreamName": "string",
    "ShardId":    number,
    "Location":   "blob",
    "Limit":      number
}

      payload = {
            "StreamName": "string",
            "ShardId":    number,
            "Location":   "blob",
            "Limit":      number
          }
      Parameters
      StreamName

      The name of the stream that contains the shard resource.

      • Type: String
      • Requirement: Required if not set in the request URL
      ShardId

      The ID of the shard from which to retrieve records.
      The shard ID is an integer between 0 and one less than the stream's shard count.

      • Type: Number
      • Requirement: Required if not set in the request URL
      Location

      The location within the shard at which to begin consuming records.

      Note
      The location must be exactly as received either in the Location element of a previous Seek response or in the NextLocation element of a previous GetRecords response. See Determining the Start Location in the operation description above. Do not attempt to calculate the location yourself.
      • Type: Blob — a Base64 encoded string
      • Requirement: Required
      Limit

      The maximum number of records to return in the response. The minimum is 1. There's no restriction on the amount of returned records, but the maximum supported overall size of all the returned records is 10 MB and the maximum size of a single record is 2 MB, so calculate the limit accordingly.

      • Type: Number
      • Requirement: Optional
      • Default Value: 1000

      Response

      Response Data

      Syntax
      {
    "NextLocation":         "blob",
    "MSecBehindLatest":     number,
    "RecordsBehindLatest":  number,
    "Records": [
        {
            "ArrivalTimeSec":   number,
            "ArrivalTimeNSec":  number,
            "SequenceNumber":   number,
            "ClientInfo":       "blob",
            "PartitionKey":     "string",
            "Data":             "blob"
        }
    ]
}
      Elements
      NextLocation

      The location of the next shard record to consume (based on the records' sequence numbers). This value should be used as the value of the Location parameter in a subsequent GetRecords request. See also Determining the Start Location in the operation description above.

      • Type: Blob — a Base64 encoded string
      MSecBehindLatest

      The difference in the ingestion time of the last record returned in the response and the latest ingested record in the shard, in milliseconds.

      • Type: Number
      RecBehindLatest

      The difference between the last record returned in the response and the latest ingested record in the shard, in number of records. A value of 0 means that the latest record in the shard has been consumed.

      • Type: Number
      Records

      An array of the requested records.

      • Type: Array of record JSON objects

      The record JSON object contains these elements:

      ArrivalTimeSec

      The record's ingestion time (the time at which the record arrived at the platform), as a Unix timestamp in seconds. For example, 1511260205 indicates that the record was ingested on 21 Nov 2017 at 10:30:05 AM. The ArrivalTimeNSec response element holds the nanoseconds unit of the ingestion time.

      • Type: Number
      ArrivalTimeNSec

      The nanoseconds unit of the ArrivalTimeSec ingestion-time timestamp. For example, if ArrivalTimeSec is 1511260205 and ArrivalTimeNSec is 500000000, the record was ingested on 21 Nov 2017 at 10:30 AM and 5.5 seconds.

      • Type: Number
      SequenceNumber

      The record's sequence number, which uniquely identifies the record within the shard. This ID number can be used in a sequence-based Seek operation to get the location of a specific record within a given stream shard.

      • Type: Number
      ClientInfo

      Custom opaque information, if provided by the producer. This metadata can be used, for example, to save the data format of a record, or the time at which a sensor or application event was triggered. See Record Metadata.

      • Type: Blob — a Base64 encoded string
      PartitionKey

      The partition key associated with the record, if provided by the producer (see the PutRecords PartitionKey record request parameter, and Record Metadata). Records with the same partition key are assigned to the same shard. See Stream Sharding and Partitioning.

      • Type: String
      Data

      Record data, as provided by the producer (see PutRecords).

      • Type: Blob — a Base64 encoded string

      Errors

      In the event of an error, the response includes a JSON object with an ErrorCode element that contains a unique numeric error code, and an ErrorMessage element that contains one of the following API error messages:
      Error Message Description
      IllegalLocation The specified record location does not exist in the shard. The requested records may have moved. Perform another Seek operation to get an updated location).
      InvalidArgumentException A provided request parameter is not valid for this request.
      Permission denied The sender of the request does not have the required permissions to perform the operation.
      ResourceNotFoundException The specified resource does not exist.
      ShardIDOutOfRangeException The specified shard does not exist in this stream.

      Examples

      Retrieve the first two records from location "AQAAAAAAAAAAAAAAAAAAAA==" in shard 199 of a MyStream stream:

      Request
        POST /mycontainer/MyDirectory/MyStream/199 HTTP/1.1
Host: https://default-tenant.app.mycluster.iguazio.com:8443
Content-Type: application/json
X-v3io-function: GetRecords
X-v3io-session-key: e8bd4ca2-537b-4175-bf01-8c74963e90bf

        {
    "Location": "AQAAAAAAAAAAAAAAAAAAAA==",
    "Limit":    2
}

        import requests

url = "https://default-tenant.app.mycluster.iguazio.com:8443/mycontainer/MyDirectory/MyStream/199"
headers = {
            "Content-Type": "application/json",
            "X-v3io-function": "GetRecords",
            "X-v3io-session-key": "e8bd4ca2-537b-4175-bf01-8c74963e90bf"
          }
payload = {"Location": "AQAAAAAAAAAAAAAAAAAAAA==", "Limit": 2}

response = requests.post(url, json=payload, headers=headers)
print(response.text)
        Response
        HTTP/1.1 200 OK
Content-Type: application/json
...

        {
    "NextLocation": "AQAAAAsAAAAFAEC2/+sAAA==",
    "MSecBehindLatest": 0,
    "RecordsBehindLatest": 0,
    "Records": [
        {
            "ArrivalTimeSec": 1485685671,
            "ArrivalTimeNSec": 160186781,
            "SequenceNumber": 15756,
            "PartitionKey": "MyKey",
            "Data": "ZGF0YQ0K"
        }
    ]
}