linar-jether

## PySpark DataFrame from many small pandas DataFrames.ipynb

      
              2 files
            
          
              4 forks
            
          
              9 comments
            
          
              4 stars
            
          
                linar-jether
                / PySpark DataFrame from many small pandas DataFrames.ipynb
            
            
              Created
              July 8, 2018 10:15
            
              
                Convert a RDD of pandas DataFrames to a single Spark DataFrame using Arrow and without collecting all data in the driver.
              
          
      Sorry, something went wrong. Reload?
      Sorry, we cannot display this file.
      Sorry, this file is invalid so it cannot be displayed.
      
          Viewer requires iframe.
      
    
## dask_celery_scheduler.py
from __future__ import absolute_import, division, print_function

import multiprocessing
import pickle
from multiprocessing.pool import ThreadPool

from celery import shared_task
from dask.local import get_async  # TODO: get better get
from dask.context import _globals
from dask.optimize import fuse, cull

## celery_task_monitor.py
import pickle
import threading
from Queue import Queue

import time
from bson import InvalidDocument

from celery.utils.log import get_task_logger

logger = get_task_logger(__name__)

## simple_python_datasource.py
from flask import Flask, request, jsonify, json, abort
from flask_cors import CORS, cross_origin

import pandas as pd

app = Flask(__name__)

cors = CORS(app)
app.config['CORS_HEADERS'] = 'Content-Type'

## dynamic_tasks.py
# Task primitives, allows pipeline execution using celery
@app.task
def dmap(it, callback, final=None):
    # Map a callback over an iterator and return as a group
    callback = subtask(callback)

    # Hack for mapping a chain to values, due to a bug where args are not copied in group creation
    if isinstance(callback, chain):
        if final:
            raise ValueError('task_processor: Cannot run reducer for dmap excecuted with a chain.')

## app_engine_remote_api.py
import getpass
import sys

sys.path.insert(0,r"C:\Program Files (x86)\Google\google_appengine")
sys.path.insert(0,r"C:\Program Files (x86)\Google\google_appengine\lib\yaml\lib")
sys.path.insert(0,r"C:\Program Files (x86)\Google\google_appengine\lib\fancy_urllib")

from google.appengine.ext.remote_api import remote_api_stub
from google.appengine.ext import db
	from __future__ import absolute_import, division, print_function

	import multiprocessing
	import pickle
	from multiprocessing.pool import ThreadPool

	from celery import shared_task
	from dask.local import get_async # TODO: get better get
	from dask.context import _globals
	from dask.optimize import fuse, cull
	import pickle
	import threading
	from Queue import Queue

	import time
	from bson import InvalidDocument

	from celery.utils.log import get_task_logger

	logger = get_task_logger(__name__)
	from flask import Flask, request, jsonify, json, abort
	from flask_cors import CORS, cross_origin

	import pandas as pd

	app = Flask(__name__)

	cors = CORS(app)
	app.config['CORS_HEADERS'] = 'Content-Type'
	# Task primitives, allows pipeline execution using celery
	@app.task
	def dmap(it, callback, final=None):
	# Map a callback over an iterator and return as a group
	callback = subtask(callback)

	# Hack for mapping a chain to values, due to a bug where args are not copied in group creation
	if isinstance(callback, chain):
	if final:
	raise ValueError('task_processor: Cannot run reducer for dmap excecuted with a chain.')
	import getpass
	import sys

	sys.path.insert(0,r"C:\Program Files (x86)\Google\google_appengine")
	sys.path.insert(0,r"C:\Program Files (x86)\Google\google_appengine\lib\yaml\lib")
	sys.path.insert(0,r"C:\Program Files (x86)\Google\google_appengine\lib\fancy_urllib")

	from google.appengine.ext.remote_api import remote_api_stub
	from google.appengine.ext import db