Radius-Kuzminki

CREATE TABLE sandbox.yurbasov_radius_kuzminki AS
-- RADIUS = 0.1 km
-- ,acos(sin((90.0 - v_gps_y) * pi() / 180.0) * sin((90.0 - nextgpsy) * pi() / 180.0)
--       * cos(radians(v_gps_x) - radians(nextgpsx)) +
--           cos((90.0 - v_gps_y) * pi() / 180.0) * cos((90.0 - nextgpsy) * pi() / 180.0)) * 6371000 as distance
-- UNITS = meters
-- Ряз 55.713745_37.809536
-- Ряз 55.726902_37.757538
-- Вол 55.703785_37.780704
-- Вол 55.70383_37.7773
-- Вол 55.704376_37.771225
-- Вол 55.704556_37.768963
-- Вол 55.70649_37.756943
-- Вол 55.70688_37.755238
WITH base1
AS (SELECT
  from_unixtime(unix_timestamp(v_time_check, 'dd.MM.yyyy HH:mm:ss'), 'yyyy-MM-dd') AS DAY,
  v_regno,
  CASE WHEN codd_kafka.v_gps_y > 55 THEN codd_kafka.v_gps_y ELSE codd_kafka.v_gps_x END AS camera_x,
  CASE WHEN codd_kafka.v_gps_y > 55 THEN codd_kafka.v_gps_x ELSE codd_kafka.v_gps_y END AS camera_y,
  lead(v_time_check, 1) OVER (partition BY v_regno ORDER BY v_time_check) AS r_previous_overall,
  v_time_check
FROM codd_data.codd codd_kafka
WHERE (from_unixtime(unix_timestamp(v_time_check, 'dd.MM.yyyy HH:mm:ss'),'yyyy-MM-dd') <= '2018-02-28'
AND from_unixtime(unix_timestamp(v_time_check, 'dd.MM.yyyy HH:mm:ss'),'yyyy-MM-dd') >= '2018-01-01')
), tgt1 AS (
SELECT 55.713745 AS v_gps_x, 37.809536 AS v_gps_y
UNION ALL SELECT 55.726902 AS v_gps_x, 37.757538 AS v_gps_y
UNION ALL SELECT 55.703785 AS v_gps_x, 37.780704 AS v_gps_y
UNION ALL SELECT 55.70383 AS v_gps_x, 37.7773 AS v_gps_y
UNION ALL SELECT 55.704376 AS v_gps_x, 37.771225 AS v_gps_y
UNION ALL SELECT 55.704556 AS v_gps_x, 37.768963 AS v_gps_y
UNION ALL SELECT 55.70649 AS v_gps_x, 37.756943 AS v_gps_y
UNION ALL SELECT 55.70688 AS v_gps_x, 37.755238 AS v_gps_y
), pre1 AS
(SELECT
  v_regno,
  camera_x, camera_y,
  MAX(abs(round((unix_timestamp(v_time_check) - unix_timestamp(r_previous_overall))/60, 0))) AS t_diff_next,
  weekofyear(v_time_check) AS w_no,
  COUNT(*) AS cnt
FROM base1
JOIN tgt1
ON acos(sin((90.0 - v_gps_y) * pi() / 180.0) * sin((90.0 - camera_y) * pi() / 180.0) * cos(radians(v_gps_x) - radians(camera_x))
       + cos((90.0 - v_gps_y) * pi() / 180.0) * cos((90.0 - camera_y) * pi() / 180.0)) * 6371000 <= 100
GROUP BY v_regno,
         camera_x, camera_y,
         weekofyear(v_time_check)),
f1 AS (SELECT pre1.v_regno, pre1.cnt FROM pre1 WHERE  t_diff_next >= 0),
xxxx1
AS (SELECT DISTINCT
  v_regno
FROM f1
WHERE cnt >= 6),
par1
AS (SELECT
  phoneno,
  carno
FROM (SELECT
  phoneno,
  carno,
  MAX(from_unixtime(unix_timestamp(parkingstart, 'dd.MM.yyyy HH:mm:ss'), 'yyyy-MM-dd')) AS max_date,
  COUNT(*) AS qty,
  ROW_NUMBER() OVER (PARTITION BY carno ORDER BY COUNT(*) DESC) AS rn
FROM parking_data.parking
WHERE phoneno != 'NULL'
AND from_unixtime(unix_timestamp(parkingstart, 'dd.MM.yyyy HH:mm:ss'), 'yyyy-MM-dd') >= '2017-01-01'
GROUP BY phoneno,
         carno) xx
WHERE rn = 1), extra1 AS (
SELECT
    xxxx1.v_regno AS v_regno
    , sec.regno_hash AS regno_hash
    , sec.regno AS regno
FROM xxxx1
 JOIN secret.regno AS sec ON xxxx1.v_regno = sec.regno_hash
 GROUP BY
 v_regno, regno_hash, regno
), res1 AS
(SELECT
  extra1.v_regno,
  extra1.regno,
  par1.phoneno
FROM extra1
JOIN par1
  ON extra1.v_regno = par1.carno
WHERE par1.carno IS NOT NULL
GROUP BY
extra1.v_regno, extra1.regno,
par1.phoneno), res AS
(SELECT res1.phoneno, res1.regno, res1.v_regno FROM res1)
SELECT s.phone, res.regno, res.v_regno FROM res JOIN secret.phones s ON s.phone_hash = res.phoneno