Соответствующие значения столбца внутри фрейма данных

Question

У меня есть датафрейм, который выглядит так:

Market         Price  date      outtime  intime  ttype
ATLJFKJFKATL   150    20190403  0215     0600    2
ATLJFK         77     20190403  0215     null    1
JFKATL         88     20190403  0600     null    1
JFKATL         77     20190403  0400     null    1

Я хотел бы взять все ttypes (roundtrip = 2, one way = 1), которые являются roundtrip (2), и сопоставить их с соответствующими путями, а затем добавить два столбца с ценой каждого. Как бы я это сделал?

Результирующий кадр данных:

Market         Price  date      outtime  intime  outbound  inbound
ATLJFKJFKATL   150    20190403  0215     0600    77        88

Это также может выглядеть так:

Market         Price  date      outtime  intime  inOutList
ATLJFKJFKATL   150    20190403  0215     0600    [77,88]

в любом случае работает. Иногда не существует единственного способа сопоставления, поэтому значение будет либо нулевым, либо пустым.

Answer 1

Вам нужно присоединиться к поездке в оба конца с одиночными поездками дважды.Ваши ключи присоединения - Market, date и time.Рынки туда и обратно должны быть разделены на 6-символьные коды, чтобы соответствовать рынкам одиночных поездок:

Сначала давайте разделим информационный кадр на отдельные поездки и туда и обратно:

import pyspark.sql.functions as psf
single, roundtrip = [df.filter(psf.col('ttype') == i).drop('ttype') for i in [1, 2]]

Чтобы извлечь исходящие и входящие рынки длятуда и обратно, мы просто используем substring:

roundtrip = roundtrip \
    .withColumn('outMarket', psf.substring('Market', 0, 6)) \
    .withColumn('inMarket', psf.substring('Market', 7, 6))

Теперь мы можем присоединиться дважды (для исходящих и входящих):

single = single \
    .drop('intime') \
    .withColumnRenamed('outtime', 'time') \
    .withColumnRenamed('Price', 'bound')
single.persist()

for bound in ['out', 'in']:
    roundtrip = roundtrip \
        .join(
            single.select([psf.col(c).alias(bound + c) for c in single.columns if c != 'date'] + ['date']), 
            on=[bound + c for c in ['Market', 'time']] + ['date'], how='left')

roundtrip.show()

        +--------+------+--------+---------+-------+------------+-----+--------+-------+
        |inMarket|intime|    date|outMarket|outtime|      Market|Price|outbound|inbound|
        +--------+------+--------+---------+-------+------------+-----+--------+-------+
        |  JFKATL|  0600|20190403|   ATLJFK|   0215|ATLJFKJFKATL|  150|      77|     88|
        +--------+------+--------+---------+-------+------------+-----+--------+-------+