MongoDB और आईटी नौकरी बाजार अनुसंधान

चुनाव Headhunter.ru पर गिर गया

उपकरण

प्रत्येक रिक्ति 30 दिनों के लिए साइट पर लटकी रहती है, जिसके बाद, यदि इसे नवीनीकृत नहीं किया जाता है, तो इसे संग्रहीत किया जाएगा। यदि 30 दिनों की समाप्ति से पहले रिक्ति को संग्रहीत किया गया था, तो इसे नियोक्ता द्वारा बंद कर दिया गया था।

हेडहंटर एपीआई (इसके बाद एचएच एपीआई के रूप में संदर्भित) आपको पिछले 30 दिनों में किसी भी तारीख के लिए प्रकाशित रिक्तियों की एक सरणी प्राप्त करने की अनुमति देता है, जिसका हम उपयोग करेंगे - हम दैनिक आधार पर प्रत्येक दिन के लिए प्रकाशित रिक्तियों को एकत्र करेंगे।

कार्यान्वयन

• कनेक्ट SQLite DB
  
  

  import sqlite3 conn_db = sqlite3.connect('hr.db', timeout=10) c = conn_db.cursor() 

• नौकरी की स्थिति में परिवर्तन के भंडारण के लिए तालिका
  सुविधा के लिए, हम SQLite डेटाबेस की एक विशेष तालिका में रिक्ति स्थिति परिवर्तन (तारीख तक उपलब्धता) के इतिहास को बचाएंगे। वेकेंसी_हॉस्टर टेबल की बदौलत, हमें साइट पर वैकेंसी की उपलब्धता के बारे में पता चल जाएगा। वह किन तिथियों में सक्रिय थी।
  
  

   c.execute(''' create table if not exists vacancy_history ( id_vacancy integer, date_load text, date_from text, date_to text )''') 

• रिक्ति छानने का काम
  एक प्रतिबंध है कि एक अनुरोध 2000 से अधिक संग्रह नहीं लौटा सकता है, और चूंकि एक दिन में साइट पर कई और रिक्तियों को प्रकाशित किया जा सकता है, इसलिए हम अनुरोध निकाय में एक फ़िल्टर डालेंगे, उदाहरण के लिए: केवल सेंट पीटर्सबर्ग में रिक्तियां (क्षेत्र = 2) , आईटी विशेषज्ञता द्वारा (विशेषज्ञता = 1)
  
  

   path = ("/vacancies?area=2&specialization=1&page={}&per_page={}&date_from={}&date_to={}".format(page, per_page, date_from, date_to)) 

• अतिरिक्त चयन शर्तें
  श्रम बाजार तेजी से बढ़ रहा है और यहां तक ​​कि फ़िल्टर को ध्यान में रखते हुए, रिक्तियों की संख्या 2000 से अधिक हो सकती है, इसलिए हम प्रत्येक दिन के लिए एक अलग लॉन्च के रूप में एक अतिरिक्त सीमा निर्धारित करेंगे: दिन के पहले आधे के लिए रिक्तियां और दूसरे छमाही के लिए रिक्तियां।
  
  

   def get_vacancy_history(): ... count_days = 30 hours = 0 while count_days >= 0: while hours < 24: date_from = (cur_date.replace(hour=hours, minute=0, second=0) - td(days=count_days)).strftime('%Y-%m-%dT%H:%M:%S') date_to = (cur_date.replace(hour=hours + 11, minute=59, second=59) - td(days=count_days)).strftime('%Y-%m-%dT%H:%M:%S') while count == per_page: path = ("/vacancies?area=2&specialization=1&page={} &per_page={}&date_from={}&date_to={}" .format(page, per_page, date_from, date_to)) conn.request("GET", path, headers=headers) response = conn.getresponse() vacancies = response.read() conn.close() count = len(json.loads(vacancies)['items']) ... #     try: c.executemany('INSERT INTO vacancy_history VALUES (?,?,?,?)', collection_for_ins) except sqlite3.DatabaseError as err: print("Error: ", err) else: conn_db.commit() if collection_for_ins: page = page + 1 total = total + count #   del(collection_for_ins[:]) hours = hours + 12 count_days = count_days - 1 hours = 0 

  
  

  c.execute(""" select a.id_vacancy, date(a.date_load) as date_last_load, date(a.date_from) as date_publish, ifnull(a.date_next, date(a.date_load, '+1 day')) as date_close from ( select vh1.id_vacancy, vh1.date_load, vh1.date_from, min(vh2.date_load) as date_next from vacancy_history vh1 left join vacancy_history vh2 on vh1.id_vacancy = vh2.id_vacancy and vh1.date_load < vh2.date_load where date(vh1.date_load) between :date_in and :date_out group by vh1.id_vacancy, vh1.date_load, vh1.date_from ) as a where a.date_next is null """, {"date_in" : date_in, "date_out" : date_out}) date_in = dt.datetime(2018, 7, 1) date_out = dt.datetime(2018, 7, 31) closed_vacancies = get_closed_by_period(date_in, date_out) df = pd.DataFrame(closed_vacancies, columns = ['id_vacancy', 'date_last_load', 'date_publish', 'date_close']) df.head() 

 #       CSV data = c.execute('select * from vacancy_history') with open('vacancy_history.csv','w', newline='') as out_csv_file: csv_out = csv.writer(out_csv_file) csv_out.writerow(d[0] for d in data.description) csv_out.writerows(data.fetchall()) conn_db.close() 

भारी तोपखाने

• मेरे MongoDB डेटाबेस का एक डेमो mLab क्लाउड सेवा में तैनात है, जो आपको मुफ्त में 500MB तक एक डेटाबेस बनाने की अनुमति देता है, जो वर्तमान कार्य को पार्स करने के लिए पर्याप्त है। Hr_db डेटाबेस में एक रिक्ति संग्रह है, जिसमें हम एक कनेक्शन स्थापित करेंगे:
  
  

   #    Mongo from pymongo import MongoClient from pymongo import ASCENDING from pymongo import errors client = MongoClient('mongodb://<db_user>:<dbpassword>@ds115219.mlab.com:15219/hr_db') db = client.hr_db VacancyMongo = db.Vacancy 

• यह ध्यान देने योग्य है कि वेतन स्तर हमेशा रूबल में इंगित नहीं किया जाता है, इसलिए, विश्लेषण के लिए सभी मूल्यों को बर्बाद करने वाले समकक्ष पर लाना आवश्यक है। ऐसा करने के लिए, हम HH API का उपयोग करते हुए शब्दकोशों का एक संग्रह पंप करते हैं, जिसमें वर्तमान तिथि के लिए विनिमय दर की जानकारी होती है:
  
  

   #   def get_dictionaries(): conn = http.client.HTTPSConnection("api.hh.ru") conn.request("GET", "https://api.hh.ru/dictionaries", headers=headers) response = conn.getresponse() if response.status != 200: conn.close() conn = http.client.HTTPSConnection("api.hh.ru") conn.request("GET", "https://api.hh.ru/dictionaries", headers=headers) response = conn.getresponse() dictionaries = response.read() dictionaries_json = json.loads(dictionaries) return dictionaries_json 

• वर्तमान विनिमय दरों के साथ मुद्राओं में शब्दकोश भरना:
  
  

   hh_dictionary = get_dictionaries() currencies = hh_dictionary['currency'] currency_rates = {} for currency in currencies: currency_rates[currency['code']] = currency['rate'] 

  
  

रिक्तियों को इकट्ठा करने के लिए उपरोक्त कार्य दैनिक आधार पर शुरू किए जाते हैं, इसलिए हर बार सभी रिक्तियों को देखने और उनमें से प्रत्येक के लिए विस्तृत जानकारी प्राप्त करने की कोई आवश्यकता नहीं है। हम केवल वही लेंगे जो पिछले पांच दिनों में प्राप्त हुए थे।

• SQLite डेटाबेस से पिछले 5 दिनों के लिए रिक्तियों की एक सरणी प्राप्त करना:
  
  

   def get_list_of_vacancies_sql(): conn_db = sqlite3.connect('hr.db', timeout=10) conn_db.row_factory = lambda cursor, row: row[0] c = conn_db.cursor() items = c.execute(""" select distinct id_vacancy from vacancy_history where date(date_load) >= date('now', '-5 day') """).fetchall() conn_db.close() return items 

• MongoDB से पिछले पांच दिनों के लिए नौकरियों की एक सरणी प्राप्त करना:
  
  

   def get_list_of_vacancies_nosql(): date_load = (dt.datetime.now() - td(days=5)).strftime('%Y-%m-%d') vacancies_from_mongo = [] for item in VacancyMongo.find({"date_load" : {"$gte" : date_load}}, {"id" : 1, "_id" : 0}): vacancies_from_mongo.append(int(item['id'])) return vacancies_from_mongo 

• यह दो सरणियों के बीच अंतर खोजने के लिए बनी हुई है, उन रिक्तियों के लिए जो मोंगोडीबी में नहीं हैं, विस्तृत जानकारी प्राप्त करें और इसे डेटाबेस में लिखें:
  
  

   sql_list = get_list_of_vacancies_sql() mongo_list = get_list_of_vacancies_nosql() vac_for_pro = [] s = set(mongo_list) vac_for_pro = [x for x in sql_list if x not in s] vac_id_chunks = [vac_for_pro[x: x + 500] for x in range(0, len(vac_for_pro), 500)] 

• इसलिए, हमारे पास नई रिक्तियों के साथ एक सरणी है जो अभी तक MongoDB में उपलब्ध नहीं हैं, उनमें से प्रत्येक के लिए हम HH API में एक अनुरोध का उपयोग करके विस्तृत जानकारी प्राप्त करेंगे, इसे सीधे MongoDB में संसाधित करने से पहले, हम प्रत्येक दस्तावेज़ को संसाधित करेंगे:
  
  1. हम रूबल के बराबर मजदूरी की राशि लाते हैं;
  2. प्रत्येक रिक्ति के लिए एक विशेषज्ञ स्तर का स्नातक जोड़ें (जूनियर / मध्य / वरिष्ठ आदि)
  
  यह सब रिक्तियों में लागू किया गया है_प्रकरण कार्य:
  
  

   from nltk.stem.snowball import SnowballStemmer stemmer = SnowballStemmer("russian") def vacancies_processing(vacancies_list): cur_date = dt.datetime.now().strftime('%Y-%m-%d') for vacancy_id in vacancies_list: conn = http.client.HTTPSConnection("api.hh.ru") conn.request("GET", "/vacancies/{}".format(vacancy_id), headers=headers) response = conn.getresponse() if response.status != 404: vacancy_txt = response.read() conn.close() vacancy = json.loads(vacancy_txt) # salary salary = None if 'salary' in vacancy: if vacancy['salary'] != None: ... max_salary = 500000 if salary is not None: salary = int(salary) if salary >= max_salary: salary = max_salary # grade grade = None if 'name' in vacancy: p_grade = '' title = re.sub(u'[^a-z-]+', ' ', vacancy['name'].lower(), re.UNICODE) words = re.split(r'\s{1,}', title.strip()) for title_word in words: title_word = stemmer.stem(title_word) if len(title_word.strip()) > 1: p_grade = p_grade + " " + title_word.strip() if re.search('()|(princip)', p_grade): grade = 'principal' elif re.search('()|(senior)|([f|F]ull)', p_grade): grade = 'senior' ... else: grade = 'not specify' vacancy['salary_processed'] = salary vacancy['date_load'] = cur_date vacancy['grade'] = grade vacancy.pop('branded_description', None) try: post_id = VacancyMongo.insert_one(vacancy) except errors.DuplicateKeyError: print ('Cant insert the duplicate vacancy_id:', vacancy['id']) 

• एचएच एपीआई तक पहुंचकर विस्तृत जानकारी प्राप्त करना, पूर्व-प्रसंस्करण प्राप्त हुआ
  MongoDB डेटा बाहर ले जाएगा और इसे कई धाराओं में सम्मिलित करेगा, प्रत्येक में 500 रिक्तियां होंगी:
  
  

   t_num = 1 threads = [] for vac_id_chunk in vac_id_chunks: print('starting', t_num) t_num = t_num + 1 t = threading.Thread(target=vacancies_processing, kwargs={'vacancies_list': vac_id_chunk}) threads.append(t) t.start() for t in threads: t.join() 

  
  

कुछ और उदाहरण

 cursor_mongo = VacancyMongo.find({"name" : {"$regex" : ".*[pP]ython*"}}) df_mongo = pd.DataFrame(list(cursor_mongo)) del df_mongo['_id'] pd.concat([df_mongo.drop(['employer'], axis=1), df_mongo['employer'].apply(pd.Series)['name']], axis=1)[['grade', 'name', 'salary_processed' ]].sort_values('salary_processed', ascending=False)[:10] 

 cursor_mongo = VacancyMongo.find({"name" : {"$regex" : ".*[jJ]ava[^sS]"}, "address" : {"$ne" : None}}) df_mongo = pd.DataFrame(list(cursor_mongo)) df_mongo['metro'] = df_mongo.apply(lambda x: x['address']['metro']['station_name'] if x['address']['metro'] is not None else None, axis = 1) df_mongo.groupby('metro')['_id'] \ .count() \ .reset_index(name='count') \ .sort_values(['count'], ascending=False) \ [:10] 

परिणाम