Saturday, October 15, 2022

Web Crawler with Redis Indexing

Summary

I'll be demonstrating a simple web crawler implementation in Nodejs in this post.  The crawled results are then stored in Redis and indexed with RediSearch.  Apache Tika is used for document parsing.

Architecture

High Level


Detailed


Crawler POST End Point


  1.  
  2. /**
  3.  * Crawl endpoint.  Starts a Worker thread (spider.js) that crawls the given fqdn, extracts text via Tika,
  4.  * and then store the resulting text in Redis as JSON documents
  5.  * This returns immediately and provides a taskID for the Worker thread. 
  6.  */
  7. app.post('/crawl', (req, res) => {
  8.     console.log(`app - POST /crawl ${req.body.fqdn}`);
  9.     const taskID = uuidv4();
  10.     try {
  11.         new Worker('./app/spider.js', { workerData : { 'fqdn': req.body.fqdn, 'taskID': taskID }});
  12.         res.status(201).json({'taskID': taskID});
  13.     }
  14.     catch (err) {
  15.         console.error(`app - POST /crawl ${req.body.fqdn} - ${err.message}`)
  16.         res.status(400).json({ 'error': err.message });
  17.     }
  18. });

Text Extraction with Tika

  1.     async extract(doc, data, hash) {
  2.         const stream = Readable.from(data);  //get a stream from the arrayBuffer obj
  3.         const response = await axios({  //send that stream to Tika for automatic mime-type detection and text extraction
  4.             method: 'PUT',
  5.             url: `${tikaUrl}/tika`,
  6.             data: stream,
  7.             responseType: 'text',
  8.             headers: {
  9.                 'Content-Type': 'application/octet-stream',
  10.                 'Accept': 'text/plain'
  11.             }
  12.         });
  13.         const json = { "doc": doc, "text": response.data, "hash": hash };
  14.         await this.client.json.set(`${PREFIX}:${doc}`, '.', json);
  15.     }

Index Creation

  1. async function buildIndex() {
  2.     console.log(`app - buildIndex`);
  3.     let rc = await clientFactory();
  4.     try {
  5.         await rc.ft.create('docIdx', {
  6.             '$.doc': {
  7.                 type: redis.SchemaFieldTypes.TEXT,
  8.                 AS: 'doc'
  9.             },
  10.             '$.text': {
  11.                 type: redis.SchemaFieldTypes.TEXT,
  12.                 AS: 'text'
  13.             }   
  14.         }, {
  15.             ON: 'JSON',
  16.             PREFIX: 'DOC'
  17.         });
  18.     }
  19.     catch(err) { 
  20.         console.error(`app - buildIndex - ${err.message}`); 
  21.     }
  22. }

Source


Copyright ©1993-2024 Joey E Whelan, All rights reserved.

Posted by at

Redis Lookaside Cache with MongoDB

Summary

In this post, I'll demonstrate how to implement caching with Redis against a Mongo database.  All code is Python using the redis-py client lib.  This is implemented as a REST API server using the Fast API module.

Architecture

High Level


Detailed


Data Set

A snippet of the data set is below

  1.  
  2. [
  3.     {
  4.         "airport": {
  5.             "code": "ATL",
  6.             "name": "Atlanta, GA: Hartsfield-Jackson Atlanta International"
  7.         },
  8.         "time": {
  9.             "label": "2003/06",
  10.             "month": 6,
  11.             "month_name": "June",
  12.             "year": 2003
  13.         },
  14.         "statistics": {
  15.             "num_delays": {
  16.                 "carrier": 1009,
  17.                 "late_aircraft": 1275,
  18.                 "national_aviation_system": 3217,
  19.                 "security": 17,
  20.                 "weather": 328
  21.             },
  22.             "carriers": {
  23.                 "names": [
  24.                     "American Airlines Inc.",
  25.                     "JetBlue Airways",
  26.                     "Continental Air Lines Inc.",
  27.                     "Delta Air Lines Inc.",
  28.                     "Atlantic Southeast Airlines",
  29.                     "AirTran Airways Corporation",
  30.                     "America West Airlines Inc.",
  31.                     "Northwest Airlines Inc.",
  32.                     "ExpressJet Airlines Inc.",
  33.                     "United Air Lines Inc.",
  34.                     "US Airways Inc."
  35.                 ],
  36.                 "total": 11
  37.             },
  38.             "flights": {
  39.                 "canceled": 216,
  40.                 "delayed": 5843,
  41.                 "diverted": 27,
  42.                 "on_time": 23974,
  43.                 "total": 30060
  44.             },
  45.             "minutes_delayed": {
  46.                 "carrier": 61606,
  47.                 "late_aircraft": 68335,
  48.                 "national_aviation_system": 118831,
  49.                 "security": 518,
  50.                 "total": 268764,
  51.                 "weather": 19474
  52.             }
  53.         }
  54.     },

Data Loading


  1.  
  2. async def startup_event():
  3.     #mongo startup
  4.     """Reset the db to an empty state and then load data from file
  5.     """
  6.     col = mdb.airlines.delays
  7.     col.drop()  
  8.     with open('data.json') as json_file:
  9.         data = load(json_file)
  10.     await col.insert_many(data)
  11.     await col.create_index([('airport.code', 1),('time.year', 1),('time.month', 1)])
  12.     
  13.     #redis startup
  14.     """Empty cache and rebuild index.
  15.     """
  16.     await rdb.flushdb()
  17.     idx_def = IndexDefinition(index_type=IndexType.JSON, prefix=['delayStat:'])
  18.     schema = (TextField('$.airport.code', as_name='airport', sortable=True),
  19.                 NumericField('$.time.year', as_name='year', sortable=True),
  20.                 NumericField('$.time.month', as_name='month', sortable=True))
  21.     await rdb.ft('idx').create_index(schema, definition=idx_def)

Data Fetch Logic


  1.  
  2. async def get_cancellations(airport: str, year: int, month: int):
  3.     try:
  4.         result, duration = await time_func(rdb.ft('idx').search,
  5.                 Query(f"(@airport:{airport}) (@year:[{year} {year}]) (@month:[{month} {month}])"))
  6.  
  7.         if len(result.docs) > 0:  # cache hit
  8.             jsonval = loads(result.docs[0].json)
  9.             metrics.incr_hits(duration)
  10.             json_res = {"result": jsonval['statistics']['flights']['canceled']}
  11.         else:  # cache miss
  12.             col = mdb.airlines.delays
  13.             lock = await lock_mgr.lock(f"lock:{airport}:{year}:{month}")  #fine-grained, distributed lock
  14.             result, duration = await time_func(col.find_one, 
  15.                 { "airport.code": airport,
  16.                     "time.year": {"$eq": year},
  17.                     "time.month": {"$eq": month}
  18.             })
  19.             metrics.incr_misses(duration)
  20.             if result:
  21.                 id = result.pop('_id')  # this field can't be serialized and needs to be removed
  22.                 await rdb.json().set(f"delayStat:{id}", '$', result)  #add val to cache and set TTL to 1 hour
  23.                 await rdb.expire(f"delayStat:{id}", 3600)
  24.                 json_res = {"result": result['statistics']['flights']['canceled']}
  25.             else: # not found in cache or db
  26.                 raise HTTPException(status_code=404, detail=f"Data not found for {airport} {year} {month}")
  27.     except Exception as err:
  28.         if type(err) == HTTPException:
  29.             raise err
  30.         else:
  31.             raise HTTPException(status_code=400, detail=str(err))
  32.     else:
  33.         return json_res
  34.     finally:
  35.         if lock and lock.valid:
  36.             await lock_mgr.unlock(lock)

Source


Copyright ©1993-2024 Joey E Whelan, All rights reserved.

Posted by at

RediSearch - ioredis client lib

Summary

I'll be demonstrating various Redis Search operations with the ioredis (Nodejs) client lib in this post.

Data Set Creation

A random set of objects are created and are then saved in Redis as hash sets.
  1.     const pipeline = client.pipeline();
  2.     for (let i=0; i < NUM; i++) {
  3.         const colors = (COLORS.sort(() => .5 - Math.random())).slice(0, Math.floor(Math.random() * COLORS.length))
  4.         const fields = {
  5.             'textField': `text${Math.floor(Math.random() * NUM)}`, 
  6.             'numericField': Math.floor(Math.random() * NUM), 
  7.             'tagField': colors
  8.         };
  9.         await pipeline.hmset(`item:${i}`, fields);
  10.     }
  11.     await pipeline.exec();

Index Creation

  1.     await client.call('FT.CREATE', 'idx', 'ON', 'HASH', 'PREFIX', '1', 'item:', 'SCHEMA', 
  2.         'textField', 'TEXT', 'SORTABLE',
  3.         'numericField', 'NUMERIC', 'SORTABLE',
  4.         'tagField', 'TAG'
  5.     );

Search Operations

  1. 	//Search for exact match in a Text field
  2.     let result = await client.call('FT.SEARCH', 'idx', '@textField:text1');
  3.  
  4.     //Search for a range in a Numeric field
  5.     result = await client.call('FT.SEARCH', 'idx', '@numericField:[1,3]');
  6.  
  7.     //Search for a match in a Tag Field
  8.     result = await client.call('FT.SEARCH', 'idx', '@tagField:{blue}');

Aggregation Operations

  1.     //Aggregate on a text field across all hashes in the index
  2.     let result = await client.call('FT.AGGREGATE', 'idx', '*', 'GROUPBY', '1', '@textField', 
  3.         'REDUCE', 'COUNT', '0', 'AS', 'CNT');
  4.         
  5.     //Search on a numeric range and then apply the SQRT function to a numeric field in the matches
  6.     const upper = Math.floor(Math.random() * NUM) + 1;
  7.     result = await client.call('FT.AGGREGATE', 'idx', `@numericField:[0,${upper}]`, 
  8.         'APPLY', 'SQRT(@numericField)', 'AS', 'SQRT');
  9.         
  10.     
  11.     //Search for logical OR of two tag values and use a CURSOR.  Limit return
  12.     //to 2 values to illustrate looping on a cursor.
  13.     result = await client.call('FT.AGGREGATE', 'idx', '@tagField:{ yellow | red }', 
  14.         'LOAD', '3', '@textField', '@numericField', '@tagField',
  15.         'WITHCURSOR', 'COUNT', '2'
  16.     );
  17.     console.log('FT.AGGREGATE idx @tagField:{ yellow | red } LOAD 3 @textField @numericField @tagField WITHCURSOR COUNT 2');
  18.     let items = result[0];
  19.     let cursor = result[1];
  20.     while (true) {
  21.         for (let item of items) {
  22.             if (Array.isArray(item)) {
  23.                 console.log(JSON.stringify(item))
  24.             }
  25.         }
  26.         if (cursor) {
  27.             result = await client.call('FT.CURSOR', 'READ', 'idx', cursor, 'COUNT', '2');
  28.             items = result[0];
  29.             cursor = result[1];
  30.         }
  31.         else {
  32.             break;
  33.         }
  34.     }

Index Alteration

  1.     await client.call('FT.ALTER', 'idx', 'SCHEMA', 'ADD', 'newField', 'TEXT', 'SORTABLE');
  2.     let result = await client.call('FT.INFO', 'idx');

Source


Copyright ©1993-2024 Joey E Whelan, All rights reserved.

Posted by at

Redis + Mongo Document (JSON/Search) Commands Comparison

Summary

This post will demonstrate equivalent Create, Read, Update and Delete (CRUD) commands in MongoDB and Redis.  I provide examples from Mongo's shell (mongosh), Redis CLI, node-redis client lib, and redis-py client lib.

Sample Data Set

Below is a snippet of the JSON document that is used in all the examples.

  1. [
  2.     { 
  3.         "item": "journal", 
  4.         "qty": 25, 
  5.         "tags": ["blank", "red"], 
  6.         "dim_cm": [ 14, 21 ], 
  7.         "size": { 
  8.             "h": 14, 
  9.             "w": 21, 
  10.             "uom": "cm"
  11.         }, 
  12.         "status": "E"  
  13.     },
  14.     { 
  15.         "item": "notebook", 
  16.         "qty": 50, 
  17.         "tags": ["red", "blank"], 
  18.         "dim_cm": [ 14, 21 ], 
  19.         "size": { 
  20.             "h": 14, 
  21.             "w": 21, 
  22.             "uom": "cm" 
  23.         }, 
  24.         "status": "E" 
  25.     },

Create Operation

Mongo Shell

  1. db.inventory.insertMany([
  2. { _id: 1, item: "journal", qty: 25, tags: ["blank", "red"], dim_cm: [ 14, 21 ], size: { h: 14, w: 21, uom: "cm" }, status: "E"  },
  3. { _id: 2, item: "notebook", qty: 50, tags: ["red", "blank"], dim_cm: [ 14, 21 ], size: { h: 14, w: 21, uom: "cm" }, status: "E" },
  4. { _id: 3, item: "paper", qty: 100, tags: ["red", "blank", "plain"], dim_cm: [ 14, 21 ], size: { h: 19, w: 22.85, uom: "cm" },status: "B"  },
  5. { _id: 4, item: "planner", qty: 75, tags: ["blank", "red"], dim_cm: [ 22.85, 30 ], status: "C" },
  6. { _id: 5, item: "postcard", qty: 45, tags: ["blue"], dim_cm: [ 10, 15.25 ], status: "D" }
  7. ])
  8.  
  9. db.inventory.createIndexes([
  10. {qty: 1},
  11. {item: 1},
  12. {tags: 1},
  13. {size: 1},
  14. {status: 1}
  15. ])

Redis CLI

  1. JSON.SET inventory:1 . '{"item":"journal","qty":25,"tags":["blank","red"],"dim_cm":[14,21],"size":{"h":14,"w":21,"uom":"cm"},"status":"E"}'
  2. JSON.SET inventory:2 . '{"item":"notebook","qty":50,"tags":["red","blank"],"dim_cm":[14,21],"size":{"h":14,"w":21,"uom":"cm"},"status":"E"}'
  3. JSON.SET inventory:3 . '{"item":"paper","qty":100,"tags":["red","blank","plain"],"dim_cm":[14,21],"size":{"h":19,"w":22.85,"uom":"cm"},"status":"B"}'
  4. JSON.SET inventory:4 . '{"item":"planner","qty":75,"tags":["blank","red"],"dim_cm":[22.85,30],"status":"C"}'
  5. JSON.SET inventory:5 . '{"item":"postcard","qty":45,"tags":["blue"],"dim_cm":[10,15.25],"status":"D"}'
  6.  
  7. FT.CREATE inventoryIdx ON JSON PREFIX 1 inventory: SCHEMA $.item AS item TEXT $.qty AS qty NUMERIC $.tags.* AS tags TAG $.dim_cm[0] AS dim_cm_0 NUMERIC $.dim_cm[1] AS dim_cm_1 NUMERIC $.status AS status TEXT $.size.h AS sizeh NUMERIC $.size.w AS sizew NUMERIC $.size.uom AS sizeuom TEXT

node-redis (Javascript)

  1. for (let item of dataset) {
  2.    await client.json.set(`inventory:${itemNum++}`, '.', item);
  3. }
  4.     
  5. await client.ft.create('inventoryIdx', {
  6.    '$.item': {
  7.       type: SchemaFieldTypes.TEXT,
  8.       AS: 'item'
  9.    },
  10.    '$.qty': {
  11.       type: SchemaFieldTypes.NUMERIC,
  12.       AS: 'qty'
  13.    },
  14.    '$.tags.*': {
  15.       type: SchemaFieldTypes.TAG,
  16.       AS: 'tags'
  17.    },
  18.    '$.dim_cm[0]': {
  19.       type: SchemaFieldTypes.NUMERIC,
  20.       AS: 'dim_cm_0'
  21.    },
  22.    '$.dim_cm[1]': {
  23.       type: SchemaFieldTypes.NUMERIC,
  24.       AS: 'dim_cm_1',
  25.    },
  26.    '$.status': {
  27.       type: SchemaFieldTypes.TEXT,
  28.       AS: 'status'
  29.    },
  30.    '$.size.h': {
  31.       type: SchemaFieldTypes.NUMERIC,
  32.       AS: 'sizeh'
  33.    },
  34.    '$.size.w': {
  35.       type: SchemaFieldTypes.NUMERIC,
  36.       AS: 'sizew'
  37.    },
  38.    '$.size.uom': {
  39.       type: SchemaFieldTypes.TEXT,
  40.       AS: 'sizeuom'
  41.    }
  42. }, {
  43.    ON: 'JSON',
  44.    PREFIX: 'inventory:'
  45. });

redis-py (Python)

  1. itemNum = 1
  2.  
  3. for dataItem in dataset:
  4.    client.json().set(f'inventory:{itemNum}', '.', dataItem)
  5.    itemNum += 1
  6.  
  7. index_def = IndexDefinition(
  8.    index_type=IndexType.JSON,
  9.         prefix=['inventory:']
  10. )
  11. schema = (  TextField('$.item', as_name='item'),
  12.             NumericField('$.qty', as_name='qty'),
  13.             TagField('$.tags.*', as_name='tags'),
  14.             NumericField('$.dim_cm[0]', as_name='dim_cm_0'),
  15.             NumericField('$.dim_cm[1]', as_name='dim_cm_1'),
  16.             TextField('$.status', as_name='status'),
  17.             NumericField('$.size.h', as_name='sizeh'),
  18.             NumericField('$.size.w', as_name='sizew'),
  19.             TextField('$.size.uom', as_name='sizeuom')
  20. )
  21. client.ft('inventoryIdx').create_index(schema,definition=index_def)

Read Operation - Compound Query

Mongo Shell

  1. db.inventory.find( {
  2.      status: "E",
  3.      $or: [ { qty: { $lt: 30 } }, { item: /^pa/ } ]
  4. } )

Redis CLI

  1. FT.SEARCH inventoryIdx '(@status:E) ((@qty:[-inf (30])|(@item:pa*))'

node-redis (Javascript)

  1. await client.ft.search('inventoryIdx', '(@status:E) ((@qty:[-inf (30])|(@item:pa*))');

redis-py (Python)

  1. client.ft('inventoryIdx').search(Query('(@status:E) ((@qty:[-inf (30])|(@item:pa*))')).docs

Update Operation

Mongo Shell

  1. db.inventory.update(
  2.    { _id: 2 },
  3.    {
  4.      $inc: { qty: 5 },
  5.      $set: {
  6.        item: "spiral notebook",
  7.      }
  8.    }
  9. )

Redis CLI

  1. MULTI
  2. JSON.NUMINCRBY inventory:2 '.qty' 5
  3. JSON.SET inventory:2 .item '"spiral notebook"'
  4. EXEC

node-redis (Javascript)

  1. await client
  2.    .multi()
  3.    .json.numIncrBy('inventory:2', '.qty', 5)
  4.    .json.set('inventory:2', '.item', 'spiral notebook' )
  5.    .exec();

redis-py (Python)

  1. pipe = client.pipeline()
  2. pipe.json().numincrby('inventory:2', '.qty', 5)
  3. pipe.json().set('inventory:2', '.item', 'spiral notebook') 
  4. pipe.execute()

Delete Operation

Mongo Shell

  1. db.inventory.deleteOne( { "_id" : 2 } )

Redis CLI

  1. JSON.DEL inventory:2

node-redis (Javascript)

  1. await client.json.del('inventory:2');

redis-py (Python)

  1. client.json().delete('inventory:2')

Source (with more examples)

https://github.com/Redislabs-Solution-Architects/docdevdemo

Copyright ©1993-2024 Joey E Whelan, All rights reserved.

Posted by at
Subscribe to: Posts (Atom)