<jupyter_start><jupyter_text>_Note: This is intended to be run as an EMR notebook_<jupyter_code>import pandas as pd
import boto3
import time
import dask.dataframe as dd
from dask_yarn import YarnCluster
from dask.distributed import Client
# Create a cluster where each worker has 1 cores and 4 GiB of memory:
cluster = YarnCluster(environment="/home/hadoop/environment.tar.gz",
                      worker_vcores = 1,
                      worker_memory = "4GiB"
                      )

# Scale cluster out to 8 such workers:
cluster.scale(8)

# Connect to the cluster (before proceeding, you should wait for workers to be registered by the dask scheduler, as below):
client = Client(cluster)
client
# We can't read a gzipped file into task directly, so we'll read it into
# pandas and then read it into a dask dataframe
df = pd.read_csv('https://landsat-pds.s3.amazonaws.com/c1/L8/scene_list.gz')
scene_list = dd.from_pandas(df, npartitions=8)

# Getting really odd behavior when reading CSV directly into dask, so read it into
# pandas first, then dask
df = pd.read_csv('https://detecting-poverty.s3.amazonaws.com/lscale/point2pathrow.csv')
pathrows = dd.from_pandas(df, npartitions=8)
# Add columns for scene date
scene_list['year'] = scene_list['acquisitionDate'].str[:4].astype(int)
scene_list['month'] = scene_list['acquisitionDate'].str[5:7].astype(int)
scene_list['day'] = scene_list['acquisitionDate'].str[8:10].astype(int)
scene_list['day_of_year'] = scene_list['month'] * scene_list['day']

pathrows['year'] = 2016  # Pathrows file should have a year column
pathrows['path'] = pathrows['path'].astype(int)
pathrows['row'] = pathrows['row'].astype(int)
pathrows['lonlat'] = pathrows['# lon'] + pathrows['lat']
# We only want scenes corresponding to a certain year
pathrows_mg = pathrows.merge(scene_list, on=['path', 'row', 'year'])

# Get min day of year for each pathrow. Can't get this to work in dask, so compute and
# convert back to dask (returns <800 rows, so not blowing up memory)
min_day = pd.DataFrame(pathrows_mg.groupby(['path', 'row'])['day_of_year'].min().compute()).reset_index()
min_day_dd = dd.from_pandas(min_day, npartitions=8)
# Get only those scenes corresponding to min day of year for which we have scene
pathrows_filt = pathrows_mg.merge(min_day_dd, on=['path', 'row', 'day_of_year'])

# Drop duplicate rows (e.g., if there are multiple scenes on same day)
pathrows_filt = pathrows_filt.drop_duplicates(subset='lonlat', ignore_index=True)
# Add columns with links to bands
pathrows_filt['link_prefix'] = pathrows_filt['download_url'].str.replace('index.html', '') \
                                + pathrows_filt['productId'] + '_'
pathrows_filt['B2_link'] = pathrows_filt['link_prefix'] + 'B2.TIF'
pathrows_filt['B3_link'] = pathrows_filt['link_prefix'] + 'B3.TIF'
pathrows_filt['B4_link'] = pathrows_filt['link_prefix'] + 'B4.TIF'

# Keep only columns we need
pathrows_filt = pathrows_filt[['# lon', 'lat', 'path', 'row',
                               'B2_link', 'B3_link', 'B4_link']]
pathrows_export = pathrows_filt.compute()
pathrows_export.head()
pathrows_export.shape
pathrows_export.iloc[0, -1]
# Export to CSV
# df.to_csv('my_file.gz', compression='gzip')
pathrows_export.to_csv('pathrow_scenelinks.zip', index=False, compression='zip')<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.1300543150271575, 'fraction_numerical': 0.011768255884127943, 'mean_word_length': 4.983754512635379, 'pattern_counts': {'":': 0, '<': 6, '<?xml version=': 0, '>': 5, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '1', 'dir': 'jupyter-structured-clean-dedup', 'id': '197032', 'license': 'MIT', 'n_tokens_mistral': 1147, 'n_tokens_neox': 1108, 'n_words': 294, 'path': 'preprocessing/get-landsat-scenes/get_landsat_scenes.ipynb', 'repo_name': 'mrmauer/detecting_poverty', 'size': '3314'}