spark.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "!cypress-kinit"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {},
   "outputs": [],
   "source": [
    "import sys\n",
    "import os\n",
    "\n",
    "sys.path.insert(0, '/usr/hdp/2.6.0.3-8/spark2/python')\n",
    "sys.path.insert(0, '/usr/hdp/2.6.0.3-8/spark2/python/lib/py4j-0.10.4-src.zip')\n",
    "\n",
    "os.environ['SPARK_HOME'] = '/usr/hdp/2.6.0.3-8/spark2/'\n",
    "os.environ['SPARK_CONF_DIR'] = '/etc/hadoop/synced_conf/spark2/'\n",
    "os.environ['PYSPARK_PYTHON'] = '/software/anaconda3/4.2.0/bin/python'\n",
    "\n",
    "import pyspark\n",
    "conf = pyspark.SparkConf()\n",
    "conf.setMaster(\"yarn\")\n",
    "conf.set(\"spark.driver.memory\",\"15g\")\n",
    "conf.set(\"spark.executor.memory\",\"60g\")\n",
    "conf.set(\"spark.executor.instances\",\"6\")\n",
    "conf.set(\"spark.executor.cores\",\"12\")\n",
    "\n",
    "sc = pyspark.SparkContext(conf=conf)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {},
   "outputs": [],
   "source": [
    "conf.set(\"spark.hadoop.validateOutputSpecs\", \"false\")\n",
    "conf.set(\"spark.serializer\", \"org.apache.spark.serializer.KryoSerializer\")\n",
    "sqlContext = pyspark.SQLContext(sc)\n",
    "#need to specify input and output\n",
    "indirectory = \"677project/data\"\n",
    "outdirectory = \"677project/output/sparkoutput\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "MapPartitionsRDD[7] at coalesce at NativeMethodAccessorImpl.java:0"
      ]
     },
     "execution_count": 30,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "readrdd = sc.newAPIHadoopFile(indirectory+\"/dump.ch2oh33_prod.lammpstrj\",\n",
    "                              'org.apache.hadoop.mapreduce.lib.input.TextInputFormat',\n",
    "                              'org.apache.hadoop.io.LongWritable',\n",
    "                              'org.apache.hadoop.io.Text',\n",
    "                              conf={'textinputformat.record.delimiter': 'ITEM: TIMESTEP'})\n",
    "readrdd.repartition(100).cache()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {},
   "outputs": [],
   "source": [
    "import string\n",
    "import math\n",
    "import re\n",
    "import numpy as np\n",
    "\n",
    "class Calculation(object):\n",
    "\n",
    "    rC = [5]                                                                                                           ################\n",
    "    rHS = [3, 4]                                                                                                            ################\n",
    "    rOS = [6]\n",
    "    rHW = [1]\n",
    "    rOW = [2]\n",
    "\n",
    "    def __init__(self, a, b, c, cosalpha, cosbeta, cosgamma):\n",
    "        self.a = a\n",
    "        self.b = b\n",
    "        self.c = c\n",
    "        self.cosalpha = cosalpha\n",
    "        self.cosbeta = cosbeta\n",
    "        self.cosgamma = cosgamma\n",
    "\n",
    "    def init(self, items):\n",
    "        temp = []\n",
    "        coords_replica = {}  # the real coords that are used, one of the 9 replica\n",
    "        x = []\n",
    "        tmp = []\n",
    "        OW = []\n",
    "        h1 = h2 = o1 = o2 = [None] * 100000   # at least 8 times larger than one box h2o atoms.\n",
    "    \n",
    "        type34 = items.type3 + items.type4\n",
    "        h1 = self.findNonperiodic (self.rHS,type34)   # item# and x, y, z for each h1 or h2 or o1 or o2\n",
    "        h2 = self.find (self.rHW,items.type1)   # item# and x, y, z for each h1 or h2 or o1 or o2\n",
    "        o1 = self.findNonperiodic (self.rOS,items.type6)   # item# and x, y, z for each h1 or h2 or o1 or o2\n",
    "        o2 = self.find (self.rOW,items.type2)   # item# and x, y, z for each h1 or h2 or o1 or o2\n",
    "        h1 = [k for k in h1 if k != None]  #h1\n",
    "        h2 = [k for k in h2 if k != None]  #h2\n",
    "        o1 = [k for k in o1 if k != None]  #o1\n",
    "        o2 = [k for k in o2 if k != None]  #o2\n",
    "    \n",
    "        for i in range(len(o1)):\n",
    "            for j in range(len(o2)):\n",
    "                do1o2 = self.dist(o1[i] , o2[j])\n",
    "                if do1o2 <= float(13.5):\n",
    "                    for l in range(len(h2)):\n",
    "                        do2h2 = self.dist(o2[j],h2[l])\n",
    "                        if do2h2<=1.2:\n",
    "                            do1h2 = self.dist(o1[i] , h2[l])\n",
    "                            A1 = self.ang( do1o2 , do1h2, do2h2)\n",
    "                            A2 = self.ang( do2h2 , do1h2, do1o2)\n",
    "                            #if A2>120 and do1h2<6.25:\n",
    "                            if A1<=30 and A2>=120 and do1h2<=6.25:\n",
    "                                for n in range(len(h2)):\n",
    "                                    if self.dist(o2[j], h2[n])<=1.2:\n",
    "                                        tmp.append((o2[j][4],h2[n][4]))\n",
    "                                        if (o2[j][4] not in coords_replica.keys()):\n",
    "                                            coords_replica[o2[j][4]] = (o2[j][1],o2[j][2],o2[j][3])  # save found O, H coords, x y z\n",
    "                                        if (h2[n][4] not in coords_replica.keys()):\n",
    "                                            coords_replica[h2[n][4]] = (h2[n][1],h2[n][2],h2[n][3])  # save found O, H coords, x y z\n",
    "                    for l in range(len(h1)):\n",
    "                        do1h1 = self.dist(o1[i],h1[l])\n",
    "                        if do1h1<=1.2:\n",
    "                            do2h1 = self.dist(o2[j] , h1[l])\n",
    "                            # print do1h1,do1o2,do2h1\n",
    "                            A1 = self.ang( do1o2 , do2h1, do1h1 )\n",
    "                            A2 = self.ang( do1h1 ,do2h1, do1o2 )\n",
    "                            #if  A2>120 and do2h1<6.25:\n",
    "                            if  A1<=30 and A2>=120 and do2h1<=6.25:\n",
    "                                for m in range(len(h2)):\n",
    "                                    if self.dist(o2[j],h2[m])<=1.2:  # output HBed Ow and corresponed Hw. At least two pairs, Ow-Hw1 AND Ow-Hw2\n",
    "                                        tmp.append((o2[j][4],h2[m][4])) # o2[j][4] is j line with item# ([4], already reordered\n",
    "                                        if (o2[j][4] not in coords_replica.keys()):\n",
    "                                            coords_replica[o2[j][4]] = (o2[j][1],o2[j][2],o2[j][3])  # save found O, H coords, x y z\n",
    "                                        if (h2[m][4] not in coords_replica.keys()):\n",
    "                                            coords_replica[h2[m][4]] = (h2[m][1],h2[m][2],h2[m][3])  # save found O, H coords, x y z\n",
    "        tmp = list(set(tmp))   # convert 2 lines(2 pairs in 1 h2o mlc: Ow-Hw1 AND Ow-Hw2) to 1 lines: Ow Hw1 Hw2, thus deletes repeating Ow\n",
    "        for p in range(len(tmp)):\n",
    "            for q in range(p+1,len(tmp)):  # set the repeated value to -1\n",
    "                if(tmp[p][0] == tmp[q][0] and tmp[p][1] == tmp[q][1]):\n",
    "                    tmp[q][0]= -1\n",
    "            for t in range(p+1, len(tmp)):\n",
    "                if(tmp[t][0]>0 and tmp[p][0] == tmp[t][0]):\n",
    "                    OW.append((tmp[p][0],tmp[p][1],tmp[t][1]))\n",
    "        #print (coords_replica)\n",
    "        #print(OW)\n",
    "        #print(len(OW))\n",
    "        return OW, coords_replica\n",
    "\n",
    "    def dist(self, X, Y):\n",
    "        x1 = X[1]\n",
    "        x2 = Y[1]\n",
    "        y1 = X[2]\n",
    "        y2 = Y[2]\n",
    "        z1 = X[3]\n",
    "        z2 = Y[3]\n",
    "        return 2*(x1-x2)*(y1-y2)*self.a*self.b*self.cosgamma + \\\n",
    "               2*(y1-y2)*(z1-z2)*self.b*self.c*self.cosalpha + \\\n",
    "               2*(x1-x2)*(z1-z2)*self.a*self.c*self.cosbeta + \\\n",
    "               (x1-x2)*(x1-x2)*self.a*self.a + \\\n",
    "               (y1-y2)*(y1-y2)*self.b*self.b + (z1-z2)*(z1-z2)*self.c*self.c\n",
    "\n",
    "    def ang(self,d1,d2,d3):# Note: square of the distances\n",
    "        return math.degrees(math.acos((d1 + d2 - d3)/(2.0 * math.sqrt(d1) * math.sqrt(d2))))\n",
    "\n",
    "    def findNonperiodic(self,y,c):\n",
    "        x = [None] * len(c)\n",
    "        for i in range(len(c)):\n",
    "            x[i] = ((c[i].atom_type,c[i].atom_x,c[i].atom_y,c[i].atom_z,c[i].atom_id))  # read the whole line and reorder each string\n",
    "        return (x)\n",
    "\n",
    "\n",
    "        #below is periodic boundary #\n",
    "    def find(self,y,c):\n",
    "        x=grab= grabxp = grabxn = grabyp = grabyn = grabxpyp = grabxnyn = grabxnyp = grabxpyn = [None] * len(c) * 9\n",
    "        for i in range(len(c)):\n",
    "            grab = ((c[i].atom_type,c[i].atom_x,c[i].atom_y,c[i].atom_z,c[i].atom_id))\n",
    "            grabxp = ((c[i].atom_type,c[i].atom_x+1,c[i].atom_y,c[i].atom_z,c[i].atom_id))\n",
    "            grabxn = ((c[i].atom_type,c[i].atom_x-1,c[i].atom_y,c[i].atom_z,c[i].atom_id))\n",
    "            grabyp = ((c[i].atom_type,c[i].atom_x,c[i].atom_y+1,c[i].atom_z,c[i].atom_id))\n",
    "            grabyn = ((c[i].atom_type,c[i].atom_x,c[i].atom_y-1,c[i].atom_z,c[i].atom_id))\n",
    "            grabxpyp = ((c[i].atom_type,c[i].atom_x+1,c[i].atom_y+1,c[i].atom_z,c[i].atom_id))\n",
    "            grabxpyn = ((c[i].atom_type,c[i].atom_x+1,c[i].atom_y-1,c[i].atom_z,c[i].atom_id))\n",
    "            grabxnyp = ((c[i].atom_type,c[i].atom_x-1,c[i].atom_y+1,c[i].atom_z,c[i].atom_id))\n",
    "            grabxnyn = ((c[i].atom_type,c[i].atom_x-1,c[i].atom_y-1,c[i].atom_z,c[i].atom_id))\n",
    "            x[9*i+0]=grab\n",
    "            x[9*i+1]=grabxp\n",
    "            x[9*i+2]=grabxn\n",
    "            x[9*i+3]=grabyp\n",
    "            x[9*i+4]=grabyn\n",
    "            x[9*i+5]=grabxpyp\n",
    "            x[9*i+6]=grabxpyn\n",
    "            x[9*i+7]=grabxnyp\n",
    "            x[9*i+8]=grabxnyn\n",
    "        return (x)\n",
    "\n",
    "    def coor(self, xlo, xhi, xy, ylo, yhi, xz, zlo, zhi, yz):  # calculate a,b,c,cosalpha,cosbeta,cosgamma\n",
    "        data = [None] * 6\n",
    "        xlo_bound = float(xlo)\n",
    "        xhi_bound = float(xhi)\n",
    "        xy =  float(xy)\n",
    "        ylo_bound =  float(ylo)\n",
    "        yhi_bound =  float(yhi)\n",
    "        xz =  float(xz)\n",
    "        zlo =  float(zlo)\n",
    "        zhi =  float(zhi)\n",
    "        yz =  float(yz)\n",
    "        xlo = xlo_bound - min(0.0,xy,xz,xy+xz)\n",
    "        xhi = xhi_bound - max(0.0,xy,xz,xy+xz)\n",
    "        ylo = ylo_bound - min(0.0,yz)\n",
    "        yhi = yhi_bound - max(0.0,yz)\n",
    "        lx = xhi-xlo\n",
    "        ly = yhi-ylo\n",
    "        lz = zhi-zlo\n",
    "        self.a = lx   #a\n",
    "        self.b = math.sqrt(ly*ly+xy*xy)   #b\n",
    "        self.c = math.sqrt(lz*lz+xz*xz+yz*yz)  #c\n",
    "        self.cosalpha = (xy*xz+ly*yz)/self.b/self.c   #cosalpha\n",
    "        self.cosbeta = xz/self.c    #cosbeta\n",
    "        self.cosgamma = xy/self.b   #cosgamma"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {},
   "outputs": [],
   "source": [
    "class Atom(object):\n",
    "\n",
    "    def __init__(self, atom_id,\n",
    "                 atom_type, atom_x, atom_y, atom_z):\n",
    "        self.atom_id = atom_id\n",
    "        self.atom_type = atom_type\n",
    "        self.atom_x = atom_x\n",
    "        self.atom_y = atom_y\n",
    "        self.atom_z = atom_z\n",
    "\n",
    "class BoxBounds(object):\n",
    "\n",
    "    def __init__(self, bounds_xyz1,\n",
    "                 bounds_xyz2, bounds_xyz3):\n",
    "        self.bounds_xyz1 = bounds_xyz1\n",
    "        self.bounds_xyz2 = bounds_xyz2\n",
    "        self.bounds_xyz3 = bounds_xyz3\n",
    "\n",
    "class Frame(object):\n",
    "    \n",
    "    def __init__(self, timestep, type1, type2, type3,\n",
    "                 type4, type5, type6, type7):\n",
    "        self.timestep = timestep\n",
    "        self.type1 = type1\n",
    "        self.type2 = type2\n",
    "        self.type3 = type3\n",
    "        self.type4 = type4\n",
    "        self.type5 = type5\n",
    "        self.type6 = type6\n",
    "        self.type7 = type7\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "import sys\n",
    "from pyspark import SparkContext, SparkConf, SQLContext\n",
    "\n",
    "\n",
    "\n",
    "def parse_data(data):\n",
    "    #reading the data in\n",
    "    #data here is a frame block\n",
    "    info = data.split(\"ITEM: NUMBER OF ATOMS\")\n",
    "    #convert to string and replace \\n\n",
    "    timestep = int(''.join(info[0]).replace(\"\\n\",\"\"))\n",
    "\n",
    "    info1 = ''.join(info[1])\n",
    "    #choose a way to separate the record\n",
    "    if \"xy xz yz\" in info1:\n",
    "        info = info1.split(\"ITEM: BOX BOUNDS xy xz yz pp pp pp\")\n",
    "    else:\n",
    "        info = info1.split(\"ITEM: BOX BOUNDS pp pp pp\")\n",
    "    \n",
    "    num_of_atoms = int(''.join(info[0].replace(\"\\n\",\"\")))\n",
    "    #box_bounds\n",
    "    info = ''.join(info[1]).split(\"ITEM: ATOMS id type xs ys zs\")\n",
    "    box_bounds = ''.join(info[0])\n",
    "    xyz = box_bounds.split(\"\\n\")\n",
    "    #skip \\n\n",
    "    xyz1 = ''.join(xyz[1]).split()\n",
    "    #if number of records is  less than 3, add one more record to default 0.0\n",
    "    if len(xyz1) < 3:\n",
    "        xyz1.append('0.0')\n",
    "    xyz2 = ''.join(xyz[2]).split()\n",
    "    if len(xyz2) < 3:\n",
    "        xyz2.append('0.0')\n",
    "    xyz3 = ''.join(xyz[3]).split()\n",
    "    if len(xyz3) < 3:\n",
    "        xyz3.append('0.0')\n",
    "    bounds = BoxBounds((float(xyz1[0]),float(xyz1[1]), float(xyz1[2])),\n",
    "                       (float(xyz2[0]),float(xyz2[1]), float(xyz2[2])),\n",
    "                       (float(xyz3[0]), float(xyz3[1]), float(xyz3[2])))\n",
    "    #atoms\n",
    "    atoms = ''.join(info[1]).split(\"\\n\")\n",
    "    #atoms_list = []\n",
    "    type1 = []\n",
    "    type2 = []\n",
    "    type3 = []\n",
    "    type4 = []\n",
    "    type5 = []\n",
    "    type6 = []\n",
    "    type7 = []\n",
    "    for atom in atoms:\n",
    "        #skip null elements\n",
    "        if atom == '':\n",
    "            continue\n",
    "        atom_elements = atom.split(\" \")\n",
    "        t_atom =Atom(int(atom_elements[0]), int(atom_elements[1]), float(atom_elements[2]),\n",
    "                               float(atom_elements[3]), float(atom_elements[4]))\n",
    "        if(t_atom.atom_type == 7): type7.append(t_atom)\n",
    "        elif(t_atom.atom_type == 2): type2.append(t_atom)\n",
    "        elif(t_atom.atom_type == 1): type1.append(t_atom)\n",
    "        elif(t_atom.atom_type == 3): type3.append(t_atom)\n",
    "        elif(t_atom.atom_type == 4): type4.append(t_atom)\n",
    "        elif(t_atom.atom_type == 5): type5.append(t_atom)\n",
    "        elif(t_atom.atom_type == 6): type6.append(t_atom)\n",
    "\n",
    "    temp_atoms = Frame(timestep, type1, type2, type3, type4, type5, type6, type7)\n",
    "    c = Calculation(0.0,0.0,0.0,0.0,0.0,0.0)#init a,b,c,cosalpha,cosbeta,cosgamma\n",
    "    xlo = bounds.bounds_xyz1[0]\n",
    "    xhi = bounds.bounds_xyz1[1]\n",
    "    xy = bounds.bounds_xyz1[2]\n",
    "    ylo = bounds.bounds_xyz2[0]\n",
    "    yhi = bounds.bounds_xyz2[1]\n",
    "    xz = bounds.bounds_xyz2[2]\n",
    "    zlo = bounds.bounds_xyz3[0]\n",
    "    zhi = bounds.bounds_xyz3[1]\n",
    "    yz = bounds.bounds_xyz3[2]\n",
    "    #do calculation\n",
    "    c.coor(xlo, xhi, xy, ylo, yhi, xz, zlo, zhi, yz)\n",
    "    hb_list, real_coords = c.init(temp_atoms)\n",
    "    hb_length = len(hb_list)\n",
    "    #print(timestep)\n",
    "    #print(\"hb_list:\", hb_list)\n",
    "    #print(\"len\",hb_length)\n",
    "    #print(\"real_coords:\", real_coords)\n",
    "    return timestep, hb_length, hb_list, real_coords\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Deleted 677project/output/sparkoutput/dump.ch2oh33_prod.lammpstrj\n",
      "1.4231519059662432\n",
      "CPU times: user 284 ms, sys: 72.5 ms, total: 356 ms\n",
      "Wall time: 16min 22s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "!hdfs dfs -rm -r -skipTrash 677project/output/sparkoutput/dump.ch2oh33_prod.lammpstrj\n",
    "frame = readrdd.values().filter(lambda x:x!=\"\").map(parse_data).cache()\n",
    "frame.saveAsTextFile(outdirectory + \"/dump.ch2oh33_prod.lammpstrj\")\n",
    "\n",
    "avgHB = frame.map(lambda x: int(x[1])).reduce(lambda x,y:x+y)/frame.count()\n",
    "#dataframe = sqlContext.createDataFrame(frame,['timestep','hb_length','hb_list','real_coords']).cache()\n",
    "#avgHB = dataframe.rdd.map(lambda x: int(x[\"hb_length\"])).reduce(lambda x,y:x+y)/dataframe.count()\n",
    "#std = dataframe.rdd.map(lambda x: int(x[\"hb_length\"])).stdev()\n",
    "\n",
    "print(\"%s\" % (avgHB))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Deleted 677project/output/sparkoutput/dump.ch2oh1212_prod.lammpstrj\r\n"
     ]
    }
   ],
   "source": [
    "!hdfs dfs -rm -r -skipTrash 677project/output/sparkoutput/dump.ch2oh1212_prod.lammpstrj"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "259.1 M  677project/output/sparkoutput/ch2oh1212_prod\r\n",
      "842.1 M  677project/output/sparkoutput/ch2oh33_prod\r\n",
      "144.4 M  677project/output/sparkoutput/ch2oh99_prod\r\n",
      "190.8 K  677project/output/sparkoutput/ch2oh_dump_1ns.lammpstrj\r\n",
      "259.1 M  677project/output/sparkoutput/dump.ch2oh1212_prod.lammpstrj\r\n",
      "842.1 M  677project/output/sparkoutput/dump.ch2oh33_prod.lammpstrj\r\n",
      "70.7 M   677project/output/sparkoutput/dump.ch2oh66_prod.lammpstrj\r\n",
      "144.4 M  677project/output/sparkoutput/dump.ch2oh99_prod.lammpstrj\r\n",
      "190.8 K  677project/output/sparkoutput/oh_1ns.lammpstrj\r\n"
     ]
    }
   ],
   "source": [
    "!hdfs dfs -du -h 677project/output/sparkoutput"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "!hdfs dfs -copyToLocal 677project/data/dump.ch2oh1212_prod.lammpstrj /scratch2/yli25"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "ch2oh.10g.lammpstrj  ch2oh.1g.lammpstrj\r\n"
     ]
    }
   ],
   "source": [
    "!ls /scratch1/xiaohoz/for_YuLi"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "!cp /scratch1/xiaohoz/for_YuLi/ch2oh.1g.lammpstrj /scratch2/yli25"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "total 47G\r\n",
      "9.7G ch2oh.10g.lammpstrj\t     28G dump.ch2oh33_prod.lammpstrj\r\n",
      "968M ch2oh.1g.lammpstrj\t\t     45K __jupyterhub__\r\n",
      "8.6G dump.ch2oh1212_prod.lammpstrj\r\n"
     ]
    }
   ],
   "source": [
    "!ls -sh /scratch2/yli25"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "6.3M ./data/ch2oh_dump_1ns.lammpstrj\r\n"
     ]
    }
   ],
   "source": [
    "!ls -sh ./data/ch2oh_dump_1ns.lammpstrj"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "913500203 /scratch2/yli25/dump.ch2oh33_prod.lammpstrj\r\n"
     ]
    }
   ],
   "source": [
    "!wc -l /scratch2/yli25/dump.ch2oh33_prod.lammpstrj"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "968M\n",
    "1.2325732484076433\n",
    "CPU times: user 147 ms, sys: 43.9 ms, total: 191 ms\n",
    "Wall time: 4min 59s\n",
    "9.7G\n",
    "1.4266803559206023\n",
    "CPU times: user 206 ms, sys: 45.5 ms, total: 252 ms\n",
    "Wall time: 7min 14s\n",
    "28G\n",
    "1.4231519059662432\n",
    "CPU times: user 284 ms, sys: 72.5 ms, total: 356 ms\n",
    "Wall time: 16min 22s"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Found 6 items\n",
      "-rw-r--r--   2 yli25 hdfs-user       9.7 G 2018-03-18 23:51 677project/data/ch2oh.10g.lammpstrj\n",
      "-rw-r--r--   2 yli25 hdfs-user     972.6 M 2018-03-18 23:46 677project/data/ch2oh.1g.lammpstrj\n",
      "-rw-r--r--   2 yli25 hdfs-user       6.2 M 2017-11-15 15:26 677project/data/ch2oh_dump_1ns.lammpstrj\n",
      "-rw-r--r--   2 yli25 hdfs-user      63.5 K 2017-11-29 15:34 677project/data/ch2oh_short.lammpstrj\n",
      "-rw-r--r--   2 yli25 hdfs-user      27.3 G 2017-11-24 22:58 677project/data/dump.ch2oh33_prod.lammpstrj\n",
      "-rw-r--r--   2 yli25 hdfs-user       4.7 M 2017-11-06 16:29 677project/data/dump.co.lammpstrj\n"
     ]
    }
   ],
   "source": [
    "!hdfs dfs -ls -h 677project/data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "(0, 2, [(76, 178, 179), (39, 104, 105)], {178: (0.306188, 0.953484, 0.233001), 179: (0.398668, 0.972962, 0.268367), 39: (0.981067, 0.356205, 0.229215), 104: (0.878979, 0.423503, 0.212988), 105: (0.897277, 0.409454, 0.250986), 76: (0.425822, 0.942096, 0.240397)})\r\n",
      "(1000, 1, [(76, 178, 179)], {178: (0.451192, 1.0727878, 0.237461), 179: (0.561458, 0.9712479, 0.272913), 76: (0.560791, 0.961417, 0.2445)})\r\n",
      "(2000, 2, [(76, 178, 179), (39, 104, 105)], {178: (0.342672, 1.00525, 0.266657), 179: (0.538719, 0.898728, 0.278641), 39: (0.9603, 0.275051, 0.228717), 104: (0.843395, 0.384878, 0.22668), 105: (1.0398646, 0.302524, 0.214019), 76: (0.455294, 0.892184, 0.261285)})\r\n",
      "(3000, 2, [(76, 178, 179), (39, 104, 105)], {178: (0.420508, 1.00257108, 0.255065), 179: (0.609856, 0.907056, 0.269455), 39: (1.00742846, 0.265535, 0.204173), 104: (0.895508, 0.381014, 0.20634), 105: (1.0426105, 0.27094, 0.178388), 76: (0.537972, 0.892895, 0.250764)})\r\n",
      "(4000, 2, [(76, 178, 179), (39, 104, 105)], {178: (0.409058, 1.0358238, 0.252193), 179: (0.609159, 0.964505, 0.245356), 39: (0.9881536, 0.204422, 0.213794), 104: (0.876908, 0.316029, 0.206985), 105: (1.0432236, 0.172618, 0.189128), 76: (0.517857, 0.926275, 0.243348)})\r\n",
      "(5000, 1, [(76, 178, 179)], {178: (0.37086, 0.99451, 0.257804), 179: (0.530927, 1.0221725, 0.25144), 76: (0.503043, 0.923176, 0.252424)})\r\n",
      "(6000, 1, [(76, 178, 179)], {178: (0.352806, 1.0820394, 0.24456), 179: (0.5517, 1.01599, 0.247165), 76: (0.468005, 0.961743, 0.24249)})\r\n",
      "(7000, 1, [(76, 178, 179)], {178: (0.470183, 1.0936832, 0.248746), 179: (0.672753, 0.970331, 0.250355), 76: (0.556982, 0.962874, 0.250211)})\r\n",
      "(8000, 1, [(76, 178, 179)], {178: (0.544653, 1.0504659, 0.249123), 179: (0.737767, 0.885357, 0.243104), 76: (0.605736, 0.914805, 0.249821)})\r\n",
      "(9000, 1, [(37, 100, 101)], {100: (0.772429, 0.492589, 0.263928), 37: (0.811488, 0.380633, 0.278604), 101: (0.697149, 0.374257, 0.280484)})\r\n",
      "(10000, 1, [(39, 104, 105)], {104: (0.534763, 1.0572115, 0.250416), 105: (0.548828, 1.0437917, 0.207767), 39: (0.566354, 0.973785, 0.231143)})\r\n",
      "(11000, 1, [(39, 104, 105)], {104: (0.254454, 1.0460188, 0.225845), 105: (0.426001, 1.0517456, 0.224948), 39: (0.380934, 0.967945, 0.221162)})\r\n",
      "(12000, 1, [(39, 104, 105)], {104: (0.492329, 1.0344606, 0.245084), 105: (0.467584, 0.919876, 0.275851), 39: (0.529925, 0.906249, 0.251995)})\r\n",
      "(13000, 1, [(39, 104, 105)], {104: (0.360279, 1.0486869, 0.221845), 105: (0.510189, 1.0192544, 0.249732), 39: (0.47434, 0.950866, 0.233236)})\r\n",
      "(14000, 1, [(39, 104, 105)], {104: (0.371249, 1.00173, 0.231516), 105: (0.477521, 0.960643, 0.266869), 39: (0.438581, 0.897676, 0.247502)})\r\n",
      "(15000, 2, [(67, 160, 161), (39, 104, 105)], {160: (0.955298, 0.619334, 0.222946), 161: (0.897729, 0.486485, 0.216585), 67: (1.00379323, 0.496512, 0.220886), 39: (0.462495, 0.946215, 0.235216), 104: (0.453828, 1.0545866, 0.226828), 105: (0.526962, 0.932709, 0.258123)})\r\n",
      "(16000, 1, [(39, 104, 105)], {104: (0.56303, 1.0697629, 0.234756), 105: (0.46838, 0.99886322, 0.258402), 39: (0.509376, 0.995468, 0.232589)})\r\n",
      "(17000, 1, [(39, 104, 105)], {104: (0.515031, 0.86676, 0.288796), 105: (0.687877, 0.871805, 0.280204), 39: (0.607569, 0.84221, 0.268665)})\r\n",
      "(18000, 1, [(39, 104, 105)], {104: (0.458646, 0.831309, 0.272085), 105: (0.595423, 0.9054, 0.273231), 39: (0.501293, 0.890407, 0.256713)})\r\n",
      "(19000, 1, [(39, 104, 105)], {104: (0.404085, 0.979124, 0.274136), 105: (0.566122, 0.807708, 0.275722), 39: (0.48461, 0.88244, 0.256773)})\r\n"
     ]
    }
   ],
   "source": [
    "!hdfs dfs -cat 677project/output/sparkoutput/oh_1ns.lammpstrj/part-00000 \\\n",
    "    2>/dev/null | head -n 20"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [],
   "source": [
    "sc.stop()"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (Anaconda)",
   "language": "python",
   "name": "anaconda_py3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.5.2"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}