Recherche sur le marché du travail MongoDB et IT

Le choix est tombé sur Headhunter.ru

Boîte à outils

Chaque poste vacant est suspendu sur le site pendant 30 jours, après quoi, s'il n'est pas renouvelé, il sera archivé. Si le poste était archivé avant l'expiration d'un délai de 30 jours, il était alors fermé par l' employeur.

L'API HeadHunter (ci-après dénommée API HH) vous permet de recevoir une série d'offres d'emploi publiées pour n'importe quelle date au cours des 30 derniers jours, que nous utiliserons - nous collecterons quotidiennement les offres publiées pour chaque jour.

Implémentation

• Connecter SQLite DB
  
  

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

• Tableau de stockage des modifications de l'état du travail
  Pour plus de commodité, nous enregistrerons l'historique du changement d'état de la vacance (disponibilité par date) dans un tableau spécial de la base de données SQLite. Grâce à la table vacancy_history, nous serons informés de la disponibilité des vacances sur le site à toute date de déchargement, c'est-à-dire à quelles dates elle était active.
  
  

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

• Filtrage des vacances
  Il y a une restriction qu'une demande ne peut pas renvoyer plus de 2000 collections, et comme il peut y avoir beaucoup plus de postes vacants publiés sur le site en une journée, nous mettrons un filtre dans le corps de la demande, par exemple: postes vacants uniquement à Saint-Pétersbourg (zone = 2) , par spécialisation informatique (spécialisation = 1)
  
  

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

• Conditions de sélection supplémentaires
  Le marché du travail croît rapidement et même en tenant compte du filtre, le nombre de postes vacants peut dépasser 2000, nous allons donc fixer une limite supplémentaire sous la forme d'un lancement distinct pour chaque jour: postes vacants pour la première moitié de la journée et postes vacants pour la seconde moitié de la journée
  
  

   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() 

Artillerie lourde

• Une démonstration de ma base de données MongoDB est déployée dans le service cloud mLab , ce qui vous permet de créer gratuitement une base de données jusqu'à 500 Mo , ce qui est suffisant pour analyser la tâche en cours. La base de données hr_db a une collection Vacancy, à laquelle nous établirons une connexion:
  
  

   #    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 

• Il convient de noter que le niveau de salaire n'est pas toujours indiqué en roubles.Par conséquent, pour l'analyse, il est nécessaire de ramener toutes les valeurs à l'équivalent en roubles. Pour ce faire, nous pompons une collection de dictionnaires à l'aide de l'API HH, qui contient des informations sur le taux de change pour la date actuelle:
  
  

   #   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 

• Remplir le dictionnaire avec des devises aux taux de change actuels:
  
  

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

  
  

Les actions susmentionnées de collecte des postes vacants sont lancées quotidiennement, il n'est donc pas nécessaire de consulter tous les postes vacants à chaque fois et de recevoir des informations détaillées pour chacun d'eux. Nous ne prendrons que ceux qui ont été reçus au cours des cinq derniers jours.

• Obtention d'un tableau d'offres d'emploi pour les 5 derniers jours à partir d'une base de données SQLite:
  
  

   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 

• Obtenir un éventail d'emplois pour les cinq derniers jours de 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 

• Il reste à trouver la différence entre les deux tableaux, pour les postes vacants qui ne sont pas dans MongoDB, obtenez des informations détaillées et écrivez-les dans la base de données:
  
  

   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)] 

• Nous avons donc un tableau avec de nouveaux postes vacants qui ne sont pas encore disponibles dans MongoDB, pour chacun d'eux, nous recevrons des informations détaillées en utilisant une demande dans l'API HH, avant de les traiter directement dans MongoDB, nous traiterons chaque document:
  
  1. Nous portons le montant du salaire à l'équivalent en roubles;
  2. Ajouter une graduation d'un niveau de spécialiste à chaque poste vacant (Junior / Moyen / Senior, etc.)
  
  Tout cela est implémenté dans la fonction vacancies_processing:
  
  

   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']) 

• Obtention d'informations détaillées en accédant à l'API HH, pré-traitement reçu
  MongoDB réalisera les données et les insérera dans plusieurs flux, avec 500 postes vacants dans chacun:
  
  

   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() 

  
  

Quelques exemples supplémentaires

 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] 

Résumé