MongoDB dan riset pasar kerja TI

Pilihan ada pada Headhunter.ru

Toolkit

Setiap lowongan menggantung di situs selama 30 hari, setelah itu, jika tidak diperbarui, itu akan diarsipkan. Jika lowongan diarsipkan sebelum berakhir 30 hari, maka ditutup oleh majikan.

HeadHunter API (selanjutnya disebut sebagai HH API) memungkinkan Anda untuk menerima serangkaian lowongan yang dipublikasikan untuk tanggal apa pun dalam 30 hari terakhir, yang akan kami gunakan - kami akan mengumpulkan lowongan yang dipublikasikan untuk setiap hari setiap hari.

Implementasi

• Hubungkan SQLite DB
  
  

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

• Tabel untuk menyimpan perubahan status pekerjaan
  Untuk kenyamanan, kami akan menyimpan riwayat perubahan status lowongan (ketersediaan berdasarkan tanggal) dalam tabel khusus dari basis data SQLite. Berkat tabel lowongan_history, kami akan mengetahui ketersediaan lowongan di situs pada setiap tanggal pengunggahan, mis. tanggal berapa dia aktif.
  
  

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

• Penyaringan kekosongan
  Ada batasan bahwa satu permintaan tidak dapat mengembalikan lebih dari 2000 koleksi, dan karena mungkin ada lebih banyak lowongan yang diterbitkan di situs dalam satu hari, kami akan meletakkan filter di badan permintaan, misalnya: lowongan hanya di St. Petersburg (area = 2) , oleh spesialisasi IT (spesialisasi = 1)
  
  

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

• Ketentuan pemilihan tambahan
  Pasar tenaga kerja tumbuh dengan cepat dan bahkan dengan mempertimbangkan filter, jumlah lowongan dapat melebihi 2000, jadi kami akan menetapkan batas tambahan dalam bentuk peluncuran terpisah untuk setiap hari: lowongan untuk paruh pertama hari itu dan lowongan untuk paruh kedua hari itu
  
  

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

Artileri berat

• Demo database MongoDB saya digunakan di layanan cloud mLab , yang memungkinkan Anda membuat database hingga 500MB secara gratis, yang cukup untuk menguraikan tugas saat ini. Database hr_db memiliki koleksi Lowongan, di mana kami akan membuat koneksi:
  
  

   #    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 

• Perlu dicatat bahwa tingkat gaji tidak selalu ditunjukkan dalam rubel, oleh karena itu, untuk analisis perlu membawa semua nilai ke setara rubel. Untuk melakukan ini, kami memompa koleksi kamus menggunakan API HH, yang berisi informasi tentang nilai tukar untuk tanggal saat ini:
  
  

   #   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 

• Mengisi kamus dengan mata uang dengan nilai tukar saat ini:
  
  

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

  
  

Tindakan yang disebutkan di atas untuk mengumpulkan lowongan diluncurkan setiap hari, sehingga tidak perlu melihat semua lowongan setiap kali dan menerima informasi terperinci untuk masing-masing lowongan. Kami hanya akan menerima yang diterima dalam lima hari terakhir.

• Mendapatkan serangkaian lowongan selama 5 hari terakhir dari database 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 

• Mendapatkan berbagai pekerjaan selama lima hari terakhir dari 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 

• Masih menemukan perbedaan antara dua array, untuk lowongan yang tidak ada di MongoDB, dapatkan informasi terperinci dan tulis ke database:
  
  

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

• Jadi, kami memiliki array dengan lowongan baru yang belum tersedia di MongoDB, untuk masing-masing kami akan menerima informasi terperinci menggunakan permintaan di API HH, sebelum memprosesnya langsung ke MongoDB, kami akan memproses setiap dokumen:
  
  1. Kami membawa jumlah upah setara dengan rubel;
  2. Tambahkan kelulusan tingkat spesialis untuk setiap lowongan (Junior / Tengah / Senior dll)
  
  Semua ini diimplementasikan dalam fungsi 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']) 

• Memperoleh informasi terperinci dengan mengakses API HH, pra-pemrosesan diterima
  MongoDB akan melaksanakan data dan memasukkannya ke dalam beberapa aliran, dengan masing-masing 500 lowongan:
  
  

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

  
  

Beberapa contoh lagi

 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] 

Ringkasan