Firehose डिलीवरी स्ट्रीम के साथ काम

AWS Kinesis और Lambda के साथ स्ट्रीमिंग डेटा

Maksim Pecherskiy

Data Engineer

स्ट्रीम बनाने के लिए तैयार

Role ARN प्राप्त करें

boto3 क्लाइंट इनिशियलाइज़ करें

import boto3

firehose = boto3.client('firehose', 
                        aws_access_key_id=AWS_KEY_ID, 
                        aws_secret_access_key=AWS_SECRET, 
                        region_name='us-east-1')

स्ट्रीम बनाएँ!

res = firehose.create_delivery_stream(

    DeliveryStreamName = "gps-delivery-stream",

    DeliveryStreamType = "DirectPut",

    S3DestinationConfiguration = {

        "RoleARN": "arn:aws:iam::0000000:role/firehoseDeliveryRole",

        "BucketARN": "arn:aws:s3:::sd-vehicle-data"

  }

)

स्ट्रीम बनाने की प्रतिक्रिया

print(res['DeliveryStreamARN'])

# नई स्ट्रीम का ARN
"arn:aws:firehose:us-east-1:0000000:deliverystream/gps-delivery-stream"

स्ट्रीम तैयार है

स्ट्रीम में लिखना

टेलीमैटिक्स हार्डवेयर

टेलीमैटिक्स डेटा भेजना

एकल रिकॉर्ड

{

    'record_id': '939ed1d1-1740-420c-8906-445278573c7f', # <-- यूनीक रिकॉर्ड आईडी

    'timestamp': '4:25:06.000', # <-- माप का समय

    'vin': '4FTEX4944AK844294', # <-- वाहन आईडी

    'lon': 106.9447146, # <-- वाहन लोकेशन longitude

    'lat': -6.3385652, # <-- वाहन लोकेशन latitude

    'speed': 25 # <-- वाहन गति

}

रिकॉर्ड आ रहे हैं

एक और उपयोग-मामला

पैटर्न

रिकॉर्ड भेजना

res = firehose.put_record(

    DeliveryStreamName='gps-delivery-stream',

    Record = {
        'Data': payload 
    }

)

रिकॉर्ड भेजना

    Record = {
        'Data': payload 
    }

रिकॉर्ड भेजना

हमारा Record कैसा दिखता है

record = {
 'record_id': '939ed1d1-1740-420c-8906-445278573c7f',
 'timestamp': '4:25:06.000','vin': '4FTEX4944AK844294',
 'lon': 106.9447146,'lat': -6.338565200000001,
 'speed': 25}

हमें क्या भेजना है (एक string)

"939ed1d1-1740-420c-8906-445278573c7f 4:25:06.000 
4FTEX4944AK844294 106.9447146 -6.338565200000001 25"

रिकॉर्ड भेजना

payload = " ".join(
    str(value) for value in record.values()
)

print(payload)

"939ed1d1-1740-420c-8906-445278573c7f 4:25:06.000 
4FTEX4944AK844294 106.9447146 -6.338565200000001 25"

सब साथ में

record = {
 'record_id': '939ed1d1-1740-420c-8906-445278573c7f',
 'timestamp': '4:25:06.000','vin': '4FTEX4944AK844294',
 'lon': 106.9447146,'lat': -6.338565200000001, 'speed': 25}

payload = " ".join(
    str(value) for value in record.values()
)

#"939ed1d1-1740-420c-8906-445278573c7f 4:25:06.000 4FTEX4944AK844294 106.9447146 -6.338565200000001 25"

सब साथ में

res = firehose.put_record(
    DeliveryStreamName='gps-delivery-stream',
    Record = {
        'Data': payload + "\n" #<-- लाइन ब्रेक!
    }
)

बने हुए फाइल

नमूना डेटा

939ed1d1-1740-420c-8906-445278573c7f 4:25:06.000 4FTEX4944AK844294 106.9447146 -6.338565200000001 25
f29a5b3d-d0fa-43c0-9e1a-e2a5cdb8be7a 8:10:47.000 3FTEX1G5XAK844393 108.58068100000001 34.79925 37
ff8e7131-408d-463b-8d07-d016419b0656 20:26:44.000 2LAXX1C8XAK844292 114.39239199999999 36.097577 90
bc75da5f-1bf6-444c-80ad-49c180e1b8de 23:16:06.000 3FTEX1G5XAK844393 -76.6990172 2.481207 40
7bdcf779-444e-4313-83da-140461933aeb 22:28:44.000 5FTEX1MAXAK844295 -47.0145295 -21.4649238 40

बने हुए फाइल

S3 क्लाइंट बनाएँ

# Create boto3 S3 client.
s3 = boto3.client('s3', 
                  aws_access_key_id=AWS_KEY_ID, 
                  aws_secret_access_key=AWS_SECRET, 
                  region_name='us-east-1')

डेटा DataFrame में पढ़ें

# Get the object from S3
obj_data = s3.get_object(Bucket='sd-vehicle-data', Key=KEY_YOU_COPIED)

# Read read the object into a DataFrame
vehicle_data = pd.read_csv(
    data['Body'], 
    delimiter = " ", 
    names=["record_id", "timestamp", "vin", "lon", "lat", "speed"]))

vehicle_data

	record_id	timestamp	vin	lon	lat	speed
0	939ed1d1...	4:25:06.000	4FTEX4944AK844294	106.945	-6.33857	25
1	f29a5b3d...	8:10:47.000	3FTEX1G5XAK844393	108.581	34.7993	37
2	ff8e7131...	20:26:44.000	2LAXX1C8XAK844292	114.392	36.0976	90
3	bc75da5f...	23:16:06.000	3FTEX1G5XAK844393	-76.699	2.48121	40
4	7bdcf779...	22:28:44.000	5FTEX1MAXAK844295	-47.0145	-21.4649	40

समीक्षा

res = firehose.create_delivery_stream(

  DeliveryStreamName =  "gps-delivery-stream",

  DeliveryStreamType = "DirectPut",

  S3DestinationConfiguration = {

    "RoleARN": "arn:aws:iam::0000000:role/firehoseDeliveryRole",

    "BucketARN": "arn:aws:s3:::sd-vehicle-data",

  }

)

समीक्षा

	record_id	timestamp	vin	lon	lat	speed
0	939ed1d1...	4:25:06.000	4FTEX4944AK844294	106.945	-6.33857	25
1	f29a5b3d...	8:10:47.000	3FTEX1G5XAK844393	108.581	34.7993	37
2	ff8e7131...	20:26:44.000	2LAXX1C8XAK844292	114.392	36.0976	90
3	bc75da5f...	23:16:06.000	3FTEX1G5XAK844393	-76.699	2.48121	40
4	7bdcf779...	22:28:44.000	5FTEX1MAXAK844295	-47.0145	-21.4649	40

अभ्यास करते हैं!

AWS Kinesis और Lambda के साथ स्ट्रीमिंग डेटा