MongoDB y la investigación de mercado de trabajo de TI

La elección recayó en Headhunter.ru

Kit de herramientas

Cada vacante permanece en el sitio durante 30 días, luego de lo cual, si no se renueva, se archivará. Si la vacante se archivó antes del vencimiento de 30 días, el empleador la cerró .

La API HeadHunter (en lo sucesivo, la API HH) le permite recibir una serie de vacantes publicadas para cualquier fecha en los últimos 30 días, que usaremos: recopilaremos las vacantes publicadas para cada día a diario .

Implementación

• Connect SQLite DB
  
  

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

• Tabla para almacenar cambios en el estado del trabajo
  Por conveniencia, guardaremos el historial de cambios de estado de vacante (disponibilidad por fecha) en una tabla especial de la base de datos SQLite. Gracias a la tabla vacancy_history, estaremos al tanto de la disponibilidad de vacantes en el sitio en cualquier fecha de descarga, es decir. en qué fechas estuvo activa.
  
  

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

• Filtrado de vacantes
  Existe una limitación en el hecho de que una solicitud no puede devolver más de 2000 colecciones, y dado que se pueden publicar muchas más vacantes en el sitio en un día, coloque un filtro en el cuerpo de la solicitud, por ejemplo: vacantes solo en San Petersburgo (área = 2) , por especialización de TI (especialización = 1)
  
  

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

• Condiciones de selección adicionales
  El mercado laboral está creciendo rápidamente e incluso teniendo en cuenta el filtro, el número de vacantes puede exceder 2000, por lo que estableceremos un límite adicional en forma de un lanzamiento separado para cada día: vacantes para la primera mitad del día y vacantes para la segunda mitad del día.
  
  

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

Artillería pesada

• Se implementa una demostración de mi base de datos MongoDB en el servicio en la nube mLab , que le permite crear una base de datos de hasta 500 MB de forma gratuita, que es suficiente para analizar la tarea actual. La base de datos hr_db tiene una colección de vacantes, a la que estableceremos una conexión:
  
  

   #    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 

• Vale la pena señalar que el nivel salarial no siempre se indica en rublos, por lo tanto, para el análisis es necesario llevar todos los valores al rublo equivalente. Para hacer esto, bombeamos una colección de diccionarios utilizando la API HH, que contiene información sobre el tipo de cambio para la fecha actual:
  
  

   #   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 

• Rellenar el diccionario con monedas con los tipos de cambio actuales:
  
  

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

  
  

Las acciones mencionadas anteriormente para la recolección de vacantes se lanzan diariamente, por lo que no es necesario ver todas las vacantes cada vez y recibir información detallada para cada una de ellas. Tomaremos solo los que fueron recibidos en los últimos cinco días.

• Obtener una variedad de vacantes para los últimos 5 días de una base de datos 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 

• Obteniendo una variedad de trabajos para los últimos cinco días 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 

• Queda por encontrar la diferencia entre las dos matrices, para aquellas vacantes que no están en MongoDB, obtenga información detallada y escríbala en la base de datos:
  
  

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

• Entonces, tenemos una matriz con nuevas vacantes que aún no están disponibles en MongoDB, recibiremos información detallada para cada uno de ellos utilizando una solicitud en la API de HH, antes de procesarla directamente en MongoDB, procesaremos cada documento:
  
  1. Traemos la cantidad de salarios al equivalente en rublos;
  2. Agregar una graduación de un nivel de especialista a cada vacante (Junior / Middle / Senior, etc.)
  
  Todo esto se implementa en la función 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']) 

• Obtención de información detallada accediendo a la API de HH, preprocesamiento recibido
  MongoDB llevará a cabo los datos y los insertará en varios flujos, con 500 vacantes en cada uno:
  
  

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

  
  

Algunos mas ejemplos

 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] 

Resumen