Как сделать мой Postgres Query быстрее, используя Apache Spark? - PullRequest
0 голосов
/ 07 марта 2019

Как минимизировать время выполнения моего запроса с помощью pyspark?

Я использую базу данных Postgres, и на моем локальном компьютере установлена ​​программа spark, имеющая 10 ГБ ОЗУ

Время выполнения запроса в PgAdmin -10 секунд

Время выполнения запроса в Pyspark - 10 секунд

Ниже приведен мой код pyspark

from pyspark.sql import DataFrameReader

url = "jdbc:postgresql://168.23.233.4:5432/MyDatabase"
properties = {
        "driver": "org.postgresql.Driver",
        "user": "postgres",
        "password": "123"
}

df = sqlContext.read.jdbc(url=url,table="(select.. very big query limit 10) AS t",     properties=properties)    
df.show()

Запрос должен присоединитьсяболее 13 таблиц каждая таблица содержит 1 миллион строк.

Пожалуйста, помогите мне ускорить запрос с помощью Spark.

Я пробовал это на основе этого блога enterОписание ссылки здесь .

Запрос Find Below, выполняющийся внутри кода pyspark,

select '2019-02-27' as "Attendance_date",e.id as e_id,concat(e.first_name::text, e.last_name::text) as "Employee_name",e.emp_id as "Employee_id",
    e.user_id as "User_id",e.customer_id,att.id attendance_id, al.id as Attendance_logs_id,aa1.id as attendance_approval_id,
    e.client_emp_id as "Client_employee_id", e.contact_no as "Contact_no", 
    att.imei as ImeiNumber,e.email_id as "Email_id",
     concat(man.first_name::text, man.last_name::text) as "Manager_name", man.id as "Manager_id",
     att.Uniform,att.Samsung_Logo,att.Blue_Color_Check,att.Blue_Color_Percentage,
    al.Face_Detection_Flag,rl.role_name as "Role_name",b.branch_name as "Branch_name",
     b.branch_code as "Branch_code",cty.city_name as "City",sm.state as "State",
     gsv1.name as "Geo_Country",gsv2.name as "Geo_State"
    ,sh.shift_name as "Shift_name",sh.id as shift_id
    ,((to_timestamp(EXTRACT(EPOCH FROM al.check_in_time::TIME) + ((tz.operator||''||tz.difference)::INTEGER))::TIME AT TIME ZONE 'utc')::TIME) 
    as "Check_in_time"
    ,al.check_in_lat as "Check_in_latitude", al.check_in_long as "Check_in_longitude",
     (select string_agg(value, ', ') from json_each_text(al.check_in_address::json))as "Check_in_address",
     att.check_in_late as "Check_in_late_remarks",al.check_in_distance_variation as "Check_in_distance",al.check_in_selfie as "Check_in_selfie",
    case when @aa1.approval_flag = 2 then ch_in.attendance_reason end as "Check_in_rejection_remarks",qc_ch_in.attendance_reason
    as "Check_in_qc_review",
    case when @att.regularize_flag = 1 or @att.approval_flag = 1 THEN 'Approved' 
    when @aa2.approval_flag = 1 THEN 'Approved' when @aa2.approval_flag = 2 THEN 'Rejected' when @aa2.approval_flag = 0 
    THEN 'Pending' else null END as "Check_out_status",
    case when att.attendance_type='P' and @att.approval_flag = 1 or att.attendance_type='L' and el.approval_flag=1 or
    att.attendance_type='H' and eh.approval_flag=1 or att.attendance_type='M' and em.approval_flag=1 or
    att.attendance_type='W' and ew.approval_flag=1 then 'Approved' 
    when att.attendance_type='P' and @att.approval_flag = 0 or att.attendance_type='L' and el.approval_flag=0 or
    att.attendance_type='H' and eh.approval_flag=0 or att.attendance_type='M' and em.approval_flag=0 or 
    att.attendance_type='W' and ew.approval_flag=0 then 'Waiting for Approval'
    when att.attendance_type='P' and el.approval_flag=2 or att.attendance_type='H' and eh.approval_flag=2 or
    att.attendance_type='M' and em.approval_flag=2 or att.attendance_type='W' and ew.approval_flag=2 then 'Rejected'
    when att.attendance_type='P' and @att.approval_flag is null then '' else 'Waiting for Approval' end as "TL approval status",
    case when att.attendance_type='P' then 'Marked' 
         when att.attendance_type='L' then 'Marked' when att.attendance_type='HL' or att.attendance_type='HP' then 'Marked' 
         when att.attendance_type='W' or ewo.weekoff_id is not null then 'Marked' when (e.customer_id is null and cehv.id is not null)
         or (e.customer_id is not null and ehv.id is not null) then 'Holiday'when el.employee_id is not null then 'Marked'
         when eh.employee_id is not null then 'Marked' when ew.employee_id is not null then 'Marked' 
         when em.employee_id is not null then 'Marked' else 'Not Marked' end "Status",
    case when att.attendance_type='P' 
    then check_in.attendance_reason when att.attendance_type='L' then lt.leave_type_name when el.employee_id is not null 
    then lt.leave_type_name when att.attendance_type='HL' or att.attendance_type='HP' then 'Half Day' 
    when att.attendance_type='W' or ewo.weekoff_id is not null then 'Week off' when (e.customer_id is null and cehv.id is not null
    ) or (e.customer_id is not null and ehv.id is not null) then 'Holiday' when eh.employee_id is not null then 'Holiday' when 
    ew.employee_id is not null then 'Week off' when em.employee_id is not null then 'Marketoff' else 'Absent' 
    end as "Attendance_reason",
    case when att.on_behalf_attendance is not null then concat(man_behalf.first_name::text,
    man_behalf.last_name::text) else null end as "Onbehalf_name",att.Check_Out_Qc_Review,att.Check_Out_Distance,
    al.Check_Out_Address
from employees e 
left join employee_applied_holidays eh on eh.employee_id=e.id and date('2019-02-27') between eh.from_date and eh.to_date 
left join employee_applied_weekoffs ew on ew.employee_id=e.id and date('2019-02-27') between ew.from_date and ew.to_date 
left join employee_applied_marketoffs em on em.employee_id=e.id and date('2019-02-27') between em.from_date and em.to_date 
inner join users u on u.ref_id = e.id and u.customer_id=200 
inner join user_role_groups urg on u.id = urg.user_id and urg.active_flag = 1
inner join attendance_setups ass on ass.role_group_id = urg.role_group_id 
left join attendances att on att.employee_id = e.id and att.start_date = '2019-02-27' and att.delete_flag = 0 

left join employee_leaves el ON el.id=(select id from employee_leaves el2 where el2.employee_id=e.id and 
el2.active_flag=1 and date('2019-02-27') between el2.from_date and el2.to_date order by id desc limit 1)

left join leave_types lt ON lt.id=(select leave_type from employee_leaves el where el.employee_id=e.id and 
el.active_flag=1 and date('2019-02-27') between el.from_date and el.to_date order by id desc limit 1)

left join attendance_logs al on al.attendance_id = att.id and al.attendance_flag = 1
left join attendance_approvals aa1 on al.id = aa1.attendance_log_id and aa1.action = 1 and aa1.active_flag = 1 
left join attendance_approvals aa2 on al.id = aa2.attendance_log_id and aa1.action = 2 and aa2.active_flag = 1 
inner join branches b on b.id = e.branch_id left join employees man on man.id = e.manager_id 
left join employees man_behalf on man_behalf.id = att.on_behalf_attendance 
left join employee_weekoff ewo on e.id = ewo.emp_id and date_part('dow','2019-02-27'::TIMESTAMP)+1 = ewo.weekoff_id and
ewo.active_flag =1 left join employee_holidays_view ehv on e.id = ehv.id and ehv.holiday_date = '2019-02-27' 
left join company_employee_holidays_view cehv on e.id = cehv.id and ehv.holiday_date = '2019-02-27' 
inner join roles rl on rl.id = e.role_id inner join cities cty on cty.id = b.city_id 
inner join states on states.id = b.state_id inner join state_master sm on sm.id = states.state_id 
inner join countries on countries.id = b.country_id inner join country_master cm on cm.country_id = countries.country_id 
left join shifts sh on sh.id = att.shift_id left join attendance_reasons ch_in on ch_in.id = aa1.reason_id 
left join sessions se on sh.id=se.shift_id left join attendance_reasons ch_out on ch_out.id = aa2.reason_id 
left join attendance_reasons qc_ch_in on qc_ch_in.id = att.check_in_qc_review 
left join attendance_reasons qc_ch_out on qc_ch_out.id = att.check_out_qc_review 
left join attendance_reasons check_in on check_in.id = al.reason_id 
left join time_zones tz on b.timezone = tz.time_zone inner join geo_outlet_mapping gom 
on b.id = gom.outlet_id left join geo_structure_values gsv1 on gsv1.id = gom.level1 left join 
geo_structure_values gsv2 on gsv2.id = gom.level2 left join geo_structure_values gsv3 on gsv3.id = gom.level3 
where e.customer_id=200

group by concat(e.first_name::text, e.last_name::text) ,e.emp_id ,e.user_id ,e.client_emp_id , e.contact_no , e.email_id,e.profile_picture,(select string_agg(role_group_name, ', ') from role_group where role_group_id = any((select array_agg(role_group_id) from user_role_groups where user_id = u.id and active_flag = 1)::int[])),concat(man.first_name::text, man.last_name::text), rl.role_name,b.branch_name,b.branch_code,cty.city_name,sm.state,cm.country,gsv1.name,gsv2.name,case when @ass.reference_point = 1 THEN b.latitude else e.latitude END,case when @ass.reference_point = 1 THEN b.longitude else e.longitude END,sh.shift_name,sh.start_time, sh.end_time,((to_timestamp(EXTRACT(EPOCH FROM al.check_in_time::TIME) + ((tz.operator||''||tz.difference)::INTEGER))::TIME AT TIME ZONE 'utc')::TIME),case 
when current_date='2019-02-27' and sh.end_time<cast(current_time as time without time zone) then null else (case when se.check_out_flag=1 then cast(att.total_hours as interval) when se.check_out_flag=0 then sh.end_time-((to_timestamp(EXTRACT(EPOCH FROM al.check_in_time::TIME) + ((tz.operator||''||tz.difference)::INTEGER))::TIME AT TIME ZONE 'utc')::TIME) end) end,al.check_in_lat, al.check_in_long,(select string_agg(value, ', ') from json_each_text(al.check_in_address::json)),att.check_in_late,al.check_in_distance_variation,al.check_in_selfie,case when @att.regularize_flag = 1 or @att.approval_flag = 1 THEN 'Approved' when @aa1.approval_flag = 1 THEN 'Approved' when @aa1.approval_flag = 2 THEN 'Rejected' when @aa1.approval_flag = 0 THEN 'Pending' else null END,case when @aa1.approval_flag = 2 then ch_in.attendance_reason end,qc_ch_in.attendance_reason,case when @att.regularize_flag = 1 or @att.approval_flag = 1 THEN 'Approved' when @aa2.approval_flag = 1 THEN 'Approved' when @aa2.approval_flag = 2 THEN 'Rejected' when @aa2.approval_flag = 0 THEN 'Pending' else null END,
    case when att.attendance_type='P' and @att.approval_flag = 1 or att.attendance_type='L' and el.approval_flag=1 or
    att.attendance_type='H' and eh.approval_flag=1 or att.attendance_type='M' and em.approval_flag=1 or
    att.attendance_type='W' and ew.approval_flag=1 then 'Approved' 
    when att.attendance_type='P' and @att.approval_flag = 0 or att.attendance_type='L' and el.approval_flag=0 or
    att.attendance_type='H' and eh.approval_flag=0 or att.attendance_type='M' and em.approval_flag=0 or 
    att.attendance_type='W' and ew.approval_flag=0 then 'Waiting for Approval'
    when att.attendance_type='P' and el.approval_flag=2 or att.attendance_type='H' and eh.approval_flag=2 or
    att.attendance_type='M' and em.approval_flag=2 or att.attendance_type='W' and ew.approval_flag=2 then 'Rejected'
    when att.attendance_type='P' and @att.approval_flag is null then '' else 'Waiting for Approval' end,
    case when att.attendance_type='P' then 'Marked' when att.attendance_type='L' then 'Marked' 
    when att.attendance_type='HL' or att.attendance_type='HP' then 'Marked' when att.attendance_type='W' 
    or ewo.weekoff_id is not null then 'Marked' when (e.customer_id is null and cehv.id is not null) or (e.customer_id is not null and ehv.id is not null) then 'Holiday' when el.employee_id is not null then 'Marked' when eh.employee_id is not null then 'Marked'  when ew.employee_id is not null then 'Marked' when em.employee_id is not null then 'Marked' else 'Not Marked' end,case when att.attendance_type='P' then check_in.attendance_reason when att.attendance_type='L' then lt.leave_type_name when el.employee_id is not null then lt.leave_type_name when att.attendance_type='HL' or att.attendance_type='HP' then 'Half Day' when att.attendance_type='W' or ewo.weekoff_id is not null then 'Week off' when (e.customer_id is null and cehv.id is not null) or (e.customer_id is not null and ehv.id is not null) then 'Holiday' when eh.employee_id is not null then 'Holiday' when ew.employee_id is not null then 'Week off' when em.employee_id is not null then 'Marketoff' else 'Absent' end ,case when att.on_behalf_attendance is not null then concat(man_behalf.first_name::text,man_behalf.last_name::text) else null end
    ,att.id,e.customer_id,al.id,aa1.id,att.imei,man.id,att.Uniform,att.Samsung_Logo,att.Blue_Color_Check,att.Blue_Color_Percentage,
    al.Face_Detection_Flag,att.Check_Out_Qc_Review,att.Check_Out_Distance,sh.id,att.id,e.id;

1 Ответ

0 голосов
/ 07 марта 2019

Насколько я знаю, в этом случае , время выполнения запроса в pyspark и pgAdmin, очевидно, займет одинаковое время, так как оба запроса выполняются поверх Только Postgres DB .

На данный момент вы еще не использовали функциональность распределенных вычислений и хранения в spark. Вы только что создали СДР на основе вывода SQL из базы данных Postgres. Только после этого ваши операции с этим СДР будут показывать разницу в скорости.

Таким образом, оптимизация будет осуществляться только на стороне Postgres DB. Ниже помогут пункты:

  1. Оптимизируйте свой SQL, чтобы он работал быстрее
  2. Считайте порции таблиц (простые SQL) в RDD и рассмотрите возможность выполнения действий / преобразований в pyspark для достижения желаемых результатов. вместо SQL со сложными объединениями.
...