[HADOOP] 어떻게 스트림으로 STDOUT 인쇄 paramiko와 맵리 듀스

어떻게 스트림으로 STDOUT 인쇄 paramiko와 맵리 듀스

나는 내가 작업을 시작 퍼티 또는 cmd를 창을 사용하지 않고 맵리 듀스 작업을 실행할 수 있습니다 paramiko를 사용하여 작은 파이썬 스크립트를 만들었습니다. 이것은 내가 작업이 완료 될 때까지 표준 출력을 볼 수 없어요 것을 제외하고 잘 작동합니다. 가 생성 될 때 나는 표준 출력의 각 라인을 볼 수 있도록 어떻게 내가 cmd를 창을 통해 할 수있을 것입니다 만,이 설정할 수 있습니다?

여기 내 스크립트입니다 :

import paramiko

# Define connection info
host_ip = 'xx.xx.xx.xx'
user = 'xxxxxxxxx'
pw = 'xxxxxxxxx'

# Commands
list_dir = "ls /nfs_home/appers/cnielsen -l"
MR = "hadoop jar /opt/cloudera/parcels/CDH/lib/hadoop-0.20-mapreduce/contrib/streaming/hadoop-streaming.jar -files /nfs_home/appers/cnielsen/product_lookups.xml -file /nfs_home/appers/cnielsen/Mapper.py -file /nfs_home/appers/cnielsen/Reducer.py -mapper '/usr/lib/python_2.7.3/bin/python Mapper.py test1' -file /nfs_home/appers/cnielsen/Process.py -reducer '/usr/lib/python_2.7.3/bin/python Reducer.py' -input /nfs_home/appers/extracts/*/*.xml -output /user/loc/output/cnielsen/test51"
getmerge = "hadoop fs -getmerge /user/loc/output/cnielsen/test51 /nfs_home/appers/cnielsen/test_010716_0.txt"

client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
client.connect(host_ip, username=user, password=pw)
##stdin, stdout, stderr = client.exec_command(list_dir)
##stdin, stdout, stderr = client.exec_command(getmerge)
stdin, stdout, stderr = client.exec_command(MR)

print "Executing command..."

for line in stdout:
    print '... ' + line.strip('\n')
for l in stderr:
    print '... ' + l.strip('\n')
client.close()

해결법

1. ==============================

   1.EOF까지이 코드는 암시 적으로 stdout.read를 호출 ()를 차단. 당신은 그러므로 즉시 출력을 얻을 덩어리에 표준 출력 / 표준 에러를 읽을해야합니다. 이 답변이 대답 특히 수정 된 버전은이 문제를 해결하는 데 도움이됩니다. 나는 몇 가지 일반적인 실속 시나리오를 방지하기 위해 사용 사례에 대한 대답이 적응하는 것이 좋습니다 것입니다.

   EOF까지이 코드는 암시 적으로 stdout.read를 호출 ()를 차단. 당신은 그러므로 즉시 출력을 얻을 덩어리에 표준 출력 / 표준 에러를 읽을해야합니다. 이 답변이 대답 특히 수정 된 버전은이 문제를 해결하는 데 도움이됩니다. 나는 몇 가지 일반적인 실속 시나리오를 방지하기 위해 사용 사례에 대한 대답이 적응하는 것이 좋습니다 것입니다.

   여기에 적응 예는 대답은 1에서 가져온 것

   sin,sout,serr = ssh.exec_command("while true; do uptime; done")
   
   def line_buffered(f):
       line_buf = ""
       while not f.channel.exit_status_ready():
           line_buf += f.read(1)
           if line_buf.endswith('\n'):
               yield line_buf
               line_buf = ''
   
   for l in line_buffered(sout):   # or serr
       print l
   

from https://stackoverflow.com/questions/34660785/mapreduce-with-paramiko-how-to-print-stdout-as-it-streams by cc-by-sa and MIT license